Функциональность mono.zip не работает должным образом - PullRequest
Новая
       6

Функциональность mono.zip не работает должным образом

0 голосов
/ 20 сентября 2018 
 Mono.zip(
            Mono.fromCallable(() -> queueDao.getNumberOfMessageInQueue(cartDaemonUrl)),
            Mono.fromCallable(() -> queueDao.getNumberOfMessageInQueue(orderConfirmationDaemonUrl))
            //ono.fromCallable(()->queueDao.getNumberOfMessageInQueue(cartDaemonUrl))
    )
            .map(T -> updateQueueCountToDb(T.getT1().block(), T.getT2().block())).
            doOnSuccess(row -> log.info("Queue info inserted into db rows ")).
            doOnError(e -> log.error("Error while inserting data stacktrace{}", e.getStackTrace()));

Я не могу понять, почему элемент управления не вводит метод updateQueueCountToDb. Я также добавил журналы в этот метод, даже эти журналы не выводятся на консоль.

1 Ответ

0 голосов
/ 20 сентября 2018

Продолжая комментарии @Jesper, которые верны, вы должны подписаться на реактивные потоки, чтобы привести их в действие, в противном случае мы просто строим реактивные конвейеры, к которым мы ожидаем, что наши данные будут поступать.

Я также изучаю реактивные потоки, поэтому ниже я создал небольшой пример, который должен вам помочь: 

Mono<String> first=Mono.just("first");
Mono<String> second=Mono.just("second");

Mono<Tuple2<String, String>> zipped=Mono.zip(first,second);

zipped.map(tuple->someOtherOperation(tuple.getT1(),tuple.getT2()))
      .doOnSuccess(s->System.out.println("Success"))
      .doOnError(s->System.out.println("Error"))
      .subscribe();

Моя реализация someOtherOperation: 

public static Mono<Void> someOtherOperation(String a, String b) 
{
    System.out.println("Performing Operation "+a+":"+b);
    return Mono.empty();
}

Итак, работаМой фрагмент кода внутри Java-приложения выведет на вашу консоль следующее: 

Performing Operation first:second
Success

Кроме того, subscribe не единственный способ подписаться на реактивные потоки, посмотрите этот документ http://projectreactor.io/docs/core/release/reference/#_simple_ways_to_create_a_flux_or_mono_and_subscribe_to_it

...