Project Reactorでページング処理を書くにはFlux#expand()を使う

Bing検索で見つけるのが難しかったのでメモ。Project Reactorでページング処理を書く方法について。

例えばこういうAPIがあったときに、どう実装するか？

class Foo { ... }

class FooPage {
  @NonNull
  Foo[] getEntities();
  Optional<Integer> getNextPageNumber();
}

interface FooRepository {
  /** @param page 0-indexed page number */
  @NonNull
  Mono<FooPage> loadPage(int page);
}

class FooService {
  @Inject // use constructor injection instead in prod.
  FooRepository repository;

  @NonNull
  Flux<Foo> loadAll() {
   // TODO
  }
}

以下のようにFlux#expand()を利用する必要があります。引数には前回ロード時の値が入っているので、そこから次回のリクエストに使用するパラメータを算出します。大抵はサーバのレスポンスに次のページ番号が入っていたり、検索パラメータとして使用すべきトークンやIDが入っているはずなので、それを引き回す形になるでしょう。

  Flux<Foo> loadAll() {
    Flux<FooPage> fluxForPage = repository.loadPage(0).expand(prevPage -> {
      return prevPage.getNextPage().map(repository::loadPage).orElse(Mono.empty());
    });
    Flux<Foo> fluxForEntity = fluxForPage.flatMap(page -> Flux.fromArray(page.getEntities()));
    return fluxForEnttiy;
  }

Reactorはリソースの開放に using() を使うのもわかりにくかったですが、ページング処理もJavadocやGitHubを検索しても出てこないのでちょっと苦労しました。