Generating a Spring Flux from a sequence of paged network calls

user_2247157 0
Viewed 1024

I am calling an API, api.magicthegathering.io/v1/cards, using the Spring reactive WebFlux client. The response is a page of 100 cards, along with headers containing links for the "next" and "last" pages, e.g. "last" is api.magicthegathering.io/v1/cards?page=426 (and "next" is simply n+1). I want to generate a Flux<Card> that feeds out each card individually, with a single entry point, e.g. Flux<Card> getAllCards().

I currently have a CardsClient component that returns a Mono<CardPage>. The CardPage has a cards() method on it that returns all cards therein (this is a 1:1 representation of the API's response model). On top of that, I have a CardCatalog component with that getAllCards() method on it.

I have tried using Flux::expand and Flux::generate, which works somewhat, but these implementations have flaws.

Here is a snippet of my current iteration of CardCatalog::getAllCards(). The problem is that the recursive nature of expand is causing redundant calls to client::getNextPage; clearly I am not using the proper method.

  @Override
  public Flux<Card> getAllCards() {
    return client.getFirstPage().flux().expand(client::getNextPage)
        .map(Page::cards)
        .flatMap(Flux::fromIterable)
        .map(mapper::convert)
        .cache();
  }

Previously I was using generate, but the issue with this is that it would always grab all pages (pretty slow), even if the subscriber only decides to take(20) cards:

 @Override
  public Flux<Card> getAllCards() {
    final Flux<Page> pageFlux =
        generate(client::getFirstPage, (response, sink) -> {
          final var page = response.block();
          sink.next(page);

          if (page.next().isPresent()) {
            return client.getNextPage(page);
          }
          sink.complete();
          return null;
        });

    return pageFlux.flatMapIterable(Page::cards).map(mapper::convert);
  }

The full code is here: https://github.com/myersadamk/mtg-api-client

Using expand, I added a print to client::getNextPage(). As you can see, the graph is creates makes redundant calls.

Getting https://api.magicthegathering.io/v1/cards?page=1
Getting https://api.magicthegathering.io/v1/cards?page=7
Getting https://api.magicthegathering.io/v1/cards?page=2
Getting https://api.magicthegathering.io/v1/cards?page=8
Getting https://api.magicthegathering.io/v1/cards?page=3
Getting https://api.magicthegathering.io/v1/cards?page=9
Getting https://api.magicthegathering.io/v1/cards?page=4
Getting https://api.magicthegathering.io/v1/cards?page=10
Getting https://api.magicthegathering.io/v1/cards?page=5
Getting https://api.magicthegathering.io/v1/cards?page=11
Getting https://api.magicthegathering.io/v1/cards?page=6
Getting https://api.magicthegathering.io/v1/cards?page=12
Getting https://api.magicthegathering.io/v1/cards?page=7

I want something more like this:

Getting https://api.magicthegathering.io/v1/cards?page=1
Getting https://api.magicthegathering.io/v1/cards?page=2
Getting https://api.magicthegathering.io/v1/cards?page=3

(Final note: it would certainly be faster to parallelize this and call the URIs directly, but it feels a little silly to bypass the next/last mechanic and hard-code the URI's. I may end up doing that, but still want to crack this nut.)

reactive spring-webflux java
2 Answers

Alright, I've come up with something that works. I decided to use the page count approach to try parallelization, although it is admittedly not faster since network IO remains the bottleneck. I'll probably go back to the header link crawling and use caching. Roughly, magic numbers and all, this is what it looks like:

  @Override
  public Flux<Card> getAllCards() {
    return client.getPageCount().flatMapMany(pageCount ->
        Flux.concat(
            range(1, pageCount)
                .parallel(pageCount / 6).runOn(Schedulers.parallel())
                .map(client::getPage)
        ).map(Page::cards).flatMap(Flux::fromIterable).map(mapper::convert)
    );
  }
user_22471570

I think this is the sequential non-blocking way of doing this:

public Flux<Card> getAllCards() {
    PaginationParams paginationParams = new PaginationParams();

    final Flux<Page> pageFlux = Mono
            .defer(() -> client.getPage(paginationParams))
            .doOnNext(page -> {
                if (page.next().isPresent()) {
                    paginationParams.setPage(page.next().get());
                } else {
                    paginationParams.setPage(null);
                }
            })
            .repeat(() -> paginationParams.getPage() != null);

    return pageFlux.flatMapIterable(Page::cards).map(mapper::convert);
}
user_145573200
Related