反应堆-使用Flux.publish（）阻止StepVerifier.thenCancel（）工作？

问题描述

如果我直接将我的助焊剂与StepVerifier一起使用， StepVerifier一切正常，但是如果我对助焊剂调用publish().autoConnect()并将结果与StepVerifier一起使用，则verify()步骤将永远不会完成。

例如，我有一个像这样的简单通量，它生成整数0到无穷大：

Flux<Integer> flux = Flux.fromStream(Stream.iterate(0, i -> i + 1));

然后，如果我仅消耗一些元素，则取消并验证，一切就可以与StepVerifier一起正常工作：

StepVerifier.create(flux)
    .thenConsumeWhile(i -> i < 10)
    .thenCancel()
    .verify();

但是，如果相反，我将执行以下操作，则它将永远不会完成：

StepVerifier.create(flux.publish().autoConnect())
    .thenConsumeWhile(i -> i < 10)
    .thenCancel()
    .verify();

我猜这里有些事情很明显，但是我不知道是什么？

当我尝试在具有如下所示的API的测试中使用StepVerifier ，这出现了：

class MyApi {
    private Flux<Integer> underlyingFlux = ...

    Flux<Integer> getFlux() { return underlyingFlux.publish().autoConnect(); }
}

请参阅我的以获取一些准备运行的代码，这些代码演示了我在StepVerifier的问题。

更新：此问题已被Reactor团队视为错误（请参阅问题 ，并且在3.2.9.RELEASE版本的Reactor-Core中提供了修复程序。