VertX / RxJava / WebClient / ApiGateway / реактивная - PullRequest
Новая
       22

VertX / RxJava / WebClient / ApiGateway / реактивная

0 голосов
/ 19 февраля 2020

Я работаю на Apigateway , используя vert.x и Rx Java. Я хочу отправить реактивный запрос на 2 Api с, получить ответ от обоих из них и отправить вместе JSON на HttpServer. Но onComplete() выполняется рано и возвращает пустое JSON. Я думаю, что проблема возникает из asynchronous символа vert.x , но я не совсем понимаю, в чем дело.

Вот мой метод: 

private void dispatchBoth(RoutingContext routingContext) {

    Observer<String> observer = new Observer<String>() {

        JsonArray jsonArray = new JsonArray();

        @Override
        public void onSubscribe(Disposable disposable) {
            System.out.println("Start");

        }

        @Override
        public void onNext(String s) {
            Thread t = new Thread(() -> {
                if(s=="/api/userApi/selectAllUsers") {

                        WebClient client = WebClient.create(vertx);
                            client
                                .get(8081, "localhost", s)
                                .send(ar->{
                                    if (ar.succeeded()) {
                                        HttpResponse<Buffer> response = ar.result();

                                        jsonArray.addAll(response.bodyAsJsonArray());
                                        System.out.println(jsonArray.encodePrettily());

                                    } else {
                                        System.out.println("Something went wrong " + ar.cause().getMessage());
                                    }
                                });

                }else if(s=="/api/holidayApi/selectAllHolidays") {
                        WebClient client = WebClient.create(vertx);
                        client
                                .get(8080, "localhost", s)
                                .send(ar -> {

                                    if (ar.succeeded()) {

                                        HttpResponse<Buffer> response = ar.result();

                                        jsonArray.addAll(response.bodyAsJsonArray());
                                       //  System.out.println(jsonArray.encodePrettily());

                                    } else {
                                        System.out.println("Something went wrong " + ar.cause().getMessage());
                                    }
                                });
                    }
                });
                t.start();
        }
        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {
                System.out.println(jsonArray.encodePrettily());
                routingContext.response().end(jsonArray.encodePrettily());
        }
    };
    Observable.fromArray(com).subscribe(observer);
}

И вот что я получаю на консоли: 

[ ]
[ {
  "holidayId" : 2,
  "userId" : 3,
  "place" : "Poland",
  "date" : {
    "year" : 2016,
    "month" : "DECEMBER",
    "dayOfMonth" : 29,
    "dayOfWeek" : "THURSDAY",
    "era" : "CE",
    "dayOfYear" : 364,
    "leapYear" : true,
    "monthValue" : 12,
    "chronology" : {
      "id" : "ISO",
      "calendarType" : "iso8601"
    }
  }
}, {
  "holidayId" : 10,
  "userId" : 1,
  "place" : "Netherland",
  "date" : {
    "year" : 2020,
    "month" : "JANUARY",
    "dayOfMonth" : 21,
    "dayOfWeek" : "TUESDAY",
    "era" : "CE",
    "dayOfYear" : 21,
    "leapYear" : true,
    "monthValue" : 1,
    "chronology" : {
      "id" : "ISO",
      "calendarType" : "iso8601"
    }
  }
}, {
  "userId" : 1,
  "name" : "Kacper",
  "phone_number" : "667667202"
}, {
  "userId" : 3,
  "name" : "Kamil",
  "phone_number" : "6734583443"
}, {
  "userId" : 4,
  "name" : "Janek",
  "phone_number" : "231253575"
}, {
  "userId" : 5,
  "name" : "Grzegorz",
  "phone_number" : "123456789"
}, {
  "userId" : 6,
  "name" : "Justin",
  "phone_number" : "111000111"
}, {
  "userId" : 8,
  "name" : "Mike",
  "phone_number" : "997"
}, {
  "userId" : 9,
  "name" : "Gorge",
  "phone_number" : "991"
} ]

Ответы [ 2 ]

0 голосов
/ 19 февраля 2020

onComplete выполняется вовремя: когда поток ввода строк завершен. Вам нужен еще один способ дождаться момента, когда все операции ввода-вывода будут завершены. Это сложная часть, я не знаю, может ли Vertx или Rx Java сделать это, но стандартный API Java может использовать CompletableFuture. Поэтому мы создаем адаптер из CompletableFuture в Handler, который содержит результат одной операции ввода-вывода: 

class HandelerFuture extends CompletableFuture<JsonArray> 
         implements Handler<AsyncResult<HttpResponse<Buffer>>> {
    @Override
    public void handle(AsyncResult<HttpResponse<Buffer>> ar) {
        if (ar.succeeded()) {
            JsonArray array = ar.result().bodyAsJsonArray();
            super.complete(array);
        } else {
            super.completeExceptionally(ar.cause());
        }
    }
}

Кроме того, вам не нужно переносить тело метода onNext в Tread. Во-вторых, вам не нужно проверять, какая строка URL передана, так как это не имеет никакого значения. В-третьих, вы используете Observer только для обработки списка строк URL-адресов - это чрезмерное усложнение. Достаточно просто l oop. 

CompletableFuture<HandelerFuture[]> dispatchBoth(String... urls) {
    ArrayList<HandelerFuture> futures = new ArrayList<>(); // all results
    for (String url : urls) {
        HandelerFuture future = new HandelerFuture();
        futures.add(future);
        WebClient client = WebClient.create(vertx);
        client.get(8081, "localhost", url)
              .send(future);
    }
    CompletableFuture all = new CompletableFuture();
    HandelerFuture[] array = futures.toArray(new HandelerFuture[0]);
    CompletableFuture.allOf(array)
            .thenRunAsync(() -> all.complete(array));
    return all;
}

, тогда его можно запустить следующим образом: 

    CompletableFuture<HandelerFuture[]> future = dispatchBoth(com);
    HandelerFuture[] results = future.get();
    JsonArray finalArray;
    for (HandelerFuture result:results) {
        try {
            // extract partial json array
            JsonArray partialArray = result.get();
            // combine partialArray with finalArray somehow
        } catch (Exception e) {
            // this is the exception got in handle() method as ar.cause().
            e.printStackTrace();
        }
    }
    routingContext.response().end(finalArray.encodePrettily());

Вы не сказали, как собираетесь объединять json массивы, поэтому Я оставил это невыполненным.

0 голосов
/ 19 февраля 2020

Я вижу, что вы создаете новую тему в onNext()? Когда вы запускаете новый поток, используя Thread.start(), он будет работать параллельно, а текущий Observable.onNext() будет считать работу завершенной. Это приведет к вызову onComplete().

Никто не ожидает завершения работы потока. Надеюсь, вы понимаете, о чем я.

Теперь я не знаком с vert.x, но, следуя указаниям Rx Java, похоже, нет необходимости использовать Thread. Вы можете удалить его и оставить остальные логики c непосредственно в методе onNext().

...