CODEWITHSUNDEEP
javaspring-webfluxproject-reactorspring-webclient
froehli
froehli

Reputation: 944

Parallel GET Request to specific mapping with WebFlux

I want to call independent request simultaneously with WebClient. My previous approch with RestTemplate was blocking my threads while waiting for the response. So I figured out, that WebClient with ParallelFlux could use one thread more efficient because it is supposed to schedule multiple requests with one thread.

My endpoint requests an tupel of id and a location.

The fooFlux method will be called a few thousand times in a loop with different parameters. The returned map will be asserted against stored reference values.

Previous attemps ofthis code resulted in duplicated API calls. But there is still a flaw. The size of the keyset of mapping is often less than the size of Set<String> location. In fact, the size of the resulting map is changing. Furthermore it is correct every now and then. So there might be an issue with the subscripton finishing after the method has returned the map.

public Map<String, ServiceDescription> fooFlux(String id, Set<String> locations) {
    Map<String, ServiceDescription> mapping = new HashMap<>();
    Flux.fromIterable(locations).parallel().runOn(Schedulers.boundedElastic()).flatMap(location -> {
        Mono<ServiceDescription> sdMono = getServiceDescription(id, location);
        Mono<Mono<ServiceDescription>> sdMonoMono = sdMono.flatMap(item -> {
            mapping.put(location, item);
            return Mono.just(sdMono);
        });
        return sdMonoMono;
    }).then().block();
    LOGGER.debug("Input Location size: {}", locations.size());
    LOGGER.debug("Output Location in map: {}", mapping.keySet().size());
    return mapping;
}

Handle Get-Request

private Mono<ServiceDescription> getServiceDescription(String id, String location) {
    String uri = URL_BASE.concat(location).concat("/detail?q=").concat(id);
    Mono<ServiceDescription> serviceDescription =
                    webClient.get().uri(uri).retrieve().onStatus(HttpStatus::isError, clientResponse -> {
                        LOGGER.error("Error while calling endpoint {} with status code {}", uri,
                                        clientResponse.statusCode());
                        throw new RuntimeException("Error while calling Endpoint");
                    }).bodyToMono(ServiceDescription.class).retryBackoff(5, Duration.ofSeconds(15));
    return serviceDescription;
}

Upvotes: 3

Views: 3853

Answers (2)

Martin Tarj&#225;nyi
Martin Tarj&#225;nyi

Reputation: 9987

public Map<String, ServiceDescription> fooFlux(String id, Set<String> locations) {
    return Flux.fromIterable(locations)
               .flatMap(location -> getServiceDescription(id, location).map(sd -> Tuples.of(location, sd)))
               .collectMap(Tuple2::getT1, Tuple2::getT2)
               .block();
}

Note: flatMap operator combined with WebClient call gives you concurrent execution, so there is no need to use ParallelFlux or any Scheduler.

Upvotes: 3

Puce
Puce

Reputation: 38152

The reactive code gets executed when you subscribe to a producer. Block does subscribe and since you call block twice (once on the Mono, but return the Mono again and then call block on the ParallelFlux), the Mono gets executed twice.

    List<String> resultList = listMono.block();
    mapping.put(location, resultList);
    return listMono;

Try something like the following instead (untested):

    listMono.map(resultList -> {
       mapping.put(location, resultList);
       return Mono.just(listMono);
    });

That said, the Reactive Programming model is quite complex, so consider to work with @Async and Future/AsyncResult instead, if this is only about calling the remote call in parallel, as others suggested. You can still use WebClient (RestTemplate seems to be on the way to get deprecated), but just call block right after bodyToMono.

Upvotes: 1

Related Questions