ProjectReactor，在开始使用 Flux 元素之前调用一个方法

Question

如何在 api 管道中调用非 Reactive 阻塞方法，但仅当 Flux 中存在至少一个项目时才调用它

例如，在调用

itemsRepo.saveItem

之前，我想调用

setupRepo

但前提是

externalApi.getAllItems()

返回的 Flux 至少发出了一个项目

externalApi
    .getAllItems()
    .flatMap(item -> 
          itemsRepo
             .saveItem(item)
             .doOnNext(e -> this.logSaveResult(item, e))
             .doOnError(th -> this.logSaveError(item, th))
    )
    .subscribe();

这里是相关的方法签名

Flux<Item> getAllItems();
Mono<Boolean> saveItem(Item item);
// The following are non reactive 
private void logSaveResult(Item item, Boolean isSuccess);
private void logSaveError(Item item, Throwable th);
void setupRepo();

如果你想测试它，这里是完整的代码。 但是这个回购协议与这个问题有一点关系。 https://github.com/mps-learning/cassandra-save