Generating paged WebClient requests and consuming response in a Flux

user_636334 0
Viewed 489

I make repeated paged WebClient requests to a third-party web service. The implementation I have now works but is blocking.

My implementation so far:

var elementsPerPage = 10;
Flux
    .generate(
        () -> 0,
        (pageIndex, emitter) -> {
            BlahServiceResponse blahServiceResponse =
                webClient
                    .get()
                    .uri("/blah?pageIndex={pageIndex}", pageIndex)
                    .retrieve()
                    .bodyToMono(BlahServiceResponse.class)
                    .block(); // Yuck!!!
            if (blahServiceResponse.getStudents().size() > 0) {
                emitter.next(blahServiceResponse);
            } else {
                emitter.complete();
            }
            return pageIndex + elementsPerPage;
        }
    )
    .subscribe(System.out::println); // Replace me with actual logic

For understandable reasons, an "IllegalStateException: The generator didn't call any of the SynchronousSink method" exception is thrown if the code above is changed to the following:

webClient
    .get()
    ...
    .bodyToMono(BlahServiceResponse.class)
    .subscribe(emitter::next);

So I started looking for an async Sink and realized it was Flux|MonoSink. But as far as I can see there is no builder method in Flux that supports stateful element generation with Flux|MonoSink.

Am I missing something and is there a more elegant approach to this?

project-reactor spring-webclient spring-webflux spring java
1 Answers

Static pagination

If you know the page indexes in advance and you have a rule to generate it.

var elementsPerPage = 10;

Flux.generate(
        () -> 0,
        (pageIndex, emitter) -> {
            if (pageIndex < 30) {
                emitter.next(pageIndex);
            } else {
                emitter.complete();
            }
            return pageIndex + elementsPerPage;
        })
        .flatMap(pageIndex -> webClient
                .get()
                .uri("/blah?pageIndex={pageIndex}", pageIndex)
                .retrieve()
                .bodyToMono(BlahServiceResponse.class))
        .subscribe(System.out::println);

Dynamic pagination

If the next page index depends on the last queried page.

public static void main(String[] args) {
    var elementsPerPage = 10;

    callWithPageIndex(0)
            .expand(pagedResponse -> {
                if (pagedResponse.getResponse().isEmpty()) {
                    return Mono.empty();
                } else {
                    return callWithPageIndex(pagedResponse.getPageIndex() + elementsPerPage);
                }
            })
            .subscribe(System.out::println);
}

private static Mono<PagedResponse<BlahServiceResponse>> callWithPageIndex(Integer pageIndex) {
    return webClient
            .get()
            .uri("/blah?pageIndex={pageIndex}", pageIndex)
            .retrieve()
            .bodyToMono(BlahServiceResponse.class)
            .map(response -> new PagedResponse<>(pageIndex, response));
}

@lombok.Value
static class PagedResponse<T> {
    int pageIndex;
    T response;
}
user_60511760
Related