如何对无限重复的Flux执行单元测试并验证首批发射元素？

问题解答 #

你只需要在现有代码后补充thenCancel().verify()即可，完整代码片段如下：

StepVerifier.
        withVirtualTime(() -> Flux.just(1,2,3,4,5).repeatWhen(c -> c.delayElements(Duration.ofSeconds(1)).log()))
        .thenAwait(Duration.ofSeconds(1))
        .expectNext(1)
        .thenAwait(Duration.ofSeconds(4))
        .expectNext(2,3,4,5)
        .thenConsumeWhile(x -> true)
        // 补充以下两行逻辑
        .thenCancel()
        .verify();

逻辑说明 #

• thenConsumeWhile(x -> true)的作用是忽略验证范围外的所有后续发射元素，避免出现「未预期的元素接收」报错
• thenCancel()会主动取消对无限Flux的订阅：因为当前流是无限重复的，永远不会发出onComplete信号，如果不主动取消，StepVerifier会一直等待流结束，最终导致测试超时失败
• verify()触发整个测试流程的执行，校验所有预设的期望是否符合实际流的行为

内容的提问来源于stack exchange，提问作者supraja