Возврат асинхронных данных ElasticSearch в сервис REST - PullRequest
Новая
       31

Возврат асинхронных данных ElasticSearch в сервис REST

0 голосов
/ 30 мая 2019

Я пытаюсь разработать ресурс REST, который перенаправляет запрос в ответственную службу по шине событий. Затем служба пытается получить некоторые данные из Elastic Search асинхронно с JavaRX.

Я использую io.reactiverse клиентскую реализацию ElasticSearch для Vert.x

Я не могу понять, как вернуть данные ElasticSearch клиенту

ElasticResource 

import io.reactivex.Single;
import io.vertx.core.json.JsonObject;
import io.vertx.reactivex.core.eventbus.EventBus;
import io.vertx.reactivex.core.eventbus.Message;
import org.elasticsearch.action.get.GetResponse;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;

@Path("/elastic")
@ApplicationScoped
public class ElasticResource {

    @Inject
    EventBus eventBus;

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    @Path("bank-es")
    public void greetingVertx(@Suspended final AsyncResponse inAsyncResponse) {

        Single<Message<GetResponse>> single = eventBus.<GetResponse>rxSend("QuarkusElasticService.getReq", new JsonObject().put("index", "bank").put("id", "1"));

        single.subscribe((mex) -> {
            inAsyncResponse.resume(Response.ok(mex.body()).build());
        });
    }
}

QuarkusElasticServiceImpl 

import com.sourcesense.sisal.socialbetting.dev.example.elastic.service.QuarkusElasticService;
import io.quarkus.vertx.ConsumeEvent;
import io.reactiverse.elasticsearch.client.reactivex.RestHighLevelClient;
import io.reactivex.Single;
import io.vertx.core.json.JsonObject;
import io.vertx.reactivex.core.Vertx;
import org.apache.http.HttpHost;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import javax.annotation.PostConstruct;
import javax.inject.Inject;
import java.util.concurrent.ExecutorService;

public class QuarkusElasticServiceImpl implements QuarkusElasticService {

    @Inject
    Vertx vertx;

    @Inject
    ExecutorService executor;

    private RestHighLevelClient esClient;

    @PostConstruct
    public void init() {
        RestClientBuilder builder = RestClient.builder(
                new HttpHost("localhost", 9200, "http"),
                new HttpHost("localhost", 9201, "http"));
        esClient = RestHighLevelClient.create(vertx, builder);

    }


    @Override
    @ConsumeEvent("QuarkusElasticService.getReq")
    public Single getReq(JsonObject jsonObject) {

        GetRequest getRequest = new GetRequest(
                jsonObject.getString("index"),
                jsonObject.getString("id"));

        return esClient.rxGetAsync(getRequest, RequestOptions.DEFAULT);
    }
}

1 Ответ

1 голос
/ 31 мая 2019

Обдумав комментарий Клемента, я нашел решение.

Прежде всего, я переключился между модулями 2 io.reactiverse, выбрав не RxJava версию io.reactiverse.elasticsearch-client.

Затем я вернулся к io.vertx.axle версиям EventBus и Message.

Затем я изменил свой код следующим образом:

ElasticResource 

import io.vertx.axle.core.eventbus.EventBus;
import io.vertx.core.json.JsonObject;

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;
import java.util.concurrent.ExecutionException;

@Path("/elastic")
@ApplicationScoped
public class ElasticResource {

    @Inject
    EventBus eventBus;

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    @Path("bank-es")
    public JsonObject greetingVertx() throws ExecutionException, InterruptedException {

        JsonObject req = new JsonObject().put("index", "bank").put("id", "1");

        return eventBus.<JsonObject>send("QuarkusElasticService.getReq", req)
                .toCompletableFuture().get().body();
    }
}

QuarkusElasticServiceImpl 

import com.sourcesense.sisal.socialbetting.dev.example.elastic.service.QuarkusElasticService;
import io.quarkus.vertx.ConsumeEvent;
import io.reactiverse.elasticsearch.client.RestHighLevelClient;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import org.apache.http.HttpHost;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import javax.annotation.PostConstruct;
import javax.inject.Inject;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

public class QuarkusElasticServiceImpl implements QuarkusElasticService {

    @Inject
    Vertx vertx;

    private RestHighLevelClient esClient;

    @PostConstruct
    public void init() {

        RestClientBuilder builder = RestClient.builder(
                new HttpHost("localhost", 9200, "http"),
                new HttpHost("localhost", 9201, "http"));
        esClient = RestHighLevelClient.create(vertx, builder);

    }


    @Override
    @ConsumeEvent("QuarkusElasticService.getReq")
    public CompletionStage<JsonObject> getReq(JsonObject jsonObject) {

        CompletableFuture future = new CompletableFuture();

        GetRequest getRequest = new GetRequest(
                jsonObject.getString("index"),
                jsonObject.getString("id"));

        esClient.getAsync(getRequest, RequestOptions.DEFAULT, ar -> {
            if (ar.failed()) {
                future.completeExceptionally(new Exception("erroraccio"));
            } else {
                future.complete(JsonObject.mapFrom(ar.result()));
            }
        });

        return future;
    }
}
...