Wait for multiple http response using reactor

Question

I'm using Spring reactor to make multiple http calls and to wait for all the results. This is my code:

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);
}

The problem is that the code above gives that Exception:

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]

I'm new in reactor developing, and I don't know why it is giving me that exception.

Could someone put me on the right way?

Answer 1

It feels like this code can be rewritten in the following manner:

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();
}