Reactor - 第一次清空时停止源

Question

我有这样的要求。

Flux<Integer> s1 = .....;

s1.flatMap(value -> anotherSource.find(value));

当anotherSource.find 给我第一个空的时候，我需要一种方法来阻止这个 s1。该怎么做？

注意：

一种可能的解决方案是抛出错误然后捕获它以停止。 anotherSource.find(value).switchIfempty(Mono.error(..))

我正在寻找比这更好的解决方案。

Answer 1

您不会为此找到特定的运算符，您必须结合运算符来实现它。 （请注意，这并不能使其本身成为“黑客”，反应式框架通常旨在以将基本运算符组合在一起以实现用例的方式使用。）

我同意使用错误来实现远非理想，因为它可能会破坏反应链中真实错误的流动 - 所以这应该是最后的手段。

如果我希望基于内部发布者停止流，我通常采用的方法是实现内部流，过滤掉onComplete() 信号，然后在适当的地方重新添加onComplete()（在这种情况下，如果它是空的。）然后您可以取消外部流的实体化，它会在您注入的任何位置响应完成的信号，停止流：

s1.flatMap(
        value ->
            anotherSource
                .find(value)
                .materialize()
                .filter(s -> !s.isOnComplete())
                .defaultIfEmpty(Signal.complete()))
    .dematerialize()

这具有保留任何错误信号的优点，同时也不需要其他对象或特殊值。