Ожидать многократного http ответа с использованием реактора - PullRequest
Новая
       62

Ожидать многократного http ответа с использованием реактора

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

Я использую Spring реактор для выполнения нескольких http-вызовов и для ожидания всех результатов.Это мой код: 

final Map<String, WSSearchResult> reduced = new HashMap<>();
List<Mono<ClientResponse>> monos = new ArrayList<>();
for (int i=0;i<10;i++) {
    log.info("Executing http call {}", i);
    WSSearchRequest wsSearchRequest = WSSearchRequest.builder().param(i).build();

    Mono<ClientResponse> exchange = webClient.post().uri("/search/availability")
            .body(BodyInserters.fromObject(wsSearchRequest)).exchange();
    monos.add(exchange);
}

final CountDownLatch latch = new CountDownLatch(monos.size());
Flux.merge(monos).subscribe(clientResponse -> {
    List<WSSearchResult> partialResult = clientResponse.bodyToFlux(WSSearchResult.class).collectList().block();
    List<WSSearchResult> partial =
            partialResult.parallelStream().filter(w-> !Strings.isNullOrEmpty(w.getId())).
                    collect(Collectors.toList());

    mapAndReduce(partial, reduced);
});

try {
    latch.await(150, TimeUnit.SECONDS);
} catch (InterruptedException e) {
    log.error(e.getMessage(), e);
}

Проблема в том, что приведенный выше код дает это исключение: 

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-6
Caused by: java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-6
    at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:77) ~[reactor-core-3.2.8.RELEASE.jar!/:3.2.8.RELEASE]
    at reactor.core.publisher.Mono.block(Mono.java:1494) ~[reactor-core-3.2.8.RELEASE.jar!/:3.2.8.RELEASE]

Я новичок в разработке реакторов, и я не знаю, почему этодает мне это исключение.

Может ли кто-нибудь поставить меня на правильный путь?

1 Ответ

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

Такое чувство, что этот код можно переписать следующим образом: 

List<Mono<ClientResponse>> clientResponses = IntStream.range(0, 10)
    .mapToObj(i -> WSSearchRequest.builder().param(i).build())
    .map(request -> send(request))
    .collect(toList());

Mono<Map<String, WSSearchResult>> reduced = Flux.merge(clientResponses)
    .flatMap(clientResponse ->
        clientResponse.bodyToFlux(WSSearchResult.class)
            .filter(result -> !Strings.isNullOrEmpty(result.getId()))
            .collectList()
            .map(listOfFilteredResults -> map(listOfFilteredResults))
    ).reduce(new HashMap<>(), (left, right) -> reduce(left, right));

reduced
  .timeout(Duration.ofSeconds(150))
  .subscribe(result -> handle(result));

private Mono<ClientResponse> send(WSSearchRequest request) {
  return webClient.post().uri("/search/availability")
      .body(BodyInserters.fromObject(wsSearchRequest))
      .exchange();
}
...