使用 StepVerifier 对 Flux.take(Duration duration) 进行单元测试

Question

我正在使用 Spring Reactor Core 3.0.6，并且我有一个返回 Flux 的方法：

public Flux<Foo> createFlux(){
    return Flux.<List<Foo>,String>generate(/* generator omitted for clarity's sake */ )
        .take(Duration.ofSeconds(10)
        .flatMap(Flux::fromIterable);
}

生成器函数调用分页的 REST api 来获取结果，如果 API 继续返回数据，我希望 Flux 只运行 10 秒。

它工作正常，但我想创建一些单元测试，但我无法创建一个测试来验证 Flux 最多只能运行 10 秒。

我模拟了其余的服务，让它总是返回数据并写了这个：

StepVerifier.withVirtualTime(() -> createFlux())
    .thenAwait(Duration.ofSeconds(10))
    .verifyComplete();

但它失败了：

java.lang.AssertionError: expectation "expectComplete" failed (expected: onComplete(); actual: onNext([my toString() Foo bean]))

我想我应该以某种方式使用生成的项目，但我找不到正确的 StepVerifier 方法来执行此操作。

编辑

我尝试使用thenConsumeWhile 跳过每个项目：

StepVerifier.withVirtualTime(() -> createFlux())
    .thenAwait(Duration.ofSeconds(10))
    .thenConsumeWhile(t -> true)
    .verifyComplete();

但现在测试只是无限期地运行并且永远不会结束。

Answer 1

生成器实际上可能非常重要...StepVerifier 受限于无限序列，在使用虚拟时间时更是如此。问题是生成器和thenAwait 都在主线程中运行，因此生成器是无限的会阻止 stepverifier 提前时间，从而防止序列超时。

由于您想测试take 的持续时间，我认为虚拟时间不合适（您正在测试时间的模拟）。我会让 createFlux 方法可以使用持续时间参数化，并执行 StepVerifier.create()，持续时间要短得多。

如果你真的想使用某种形式的虚拟时间，我发现让它工作的最低要求是

1. 通过在测试开始时实例化Scheduler 来隔离非虚拟线程上的生成器循环，然后在StepVerifier 的Supplier 中使用subscribeOn(scheduler)。
2. 首先调用.expectNextCount(1)，确保在尝试提前时间之前订阅所有内容并且数据开始流动。

像这样：

public Flux<Integer> createFlux() {
    return Flux.<List<Integer>>generate(sink -> {
        sink.next(Arrays.asList(1, 2, 3));
    })
            .take(Duration.ofSeconds(10))
            .flatMap(Flux::fromIterable);
}

@Test
public void so44657525() throws InterruptedException {
    Scheduler scheduler = Schedulers.newSingle("test");
    AtomicInteger adder = new AtomicInteger();

    StepVerifier.withVirtualTime(() -> createFlux()
            .subscribeOn(scheduler)
            .doOnNext(v -> adder.incrementAndGet())
    )
                .expectNextCount(1)
                .thenAwait(Duration.ofSeconds(10))
                .thenConsumeWhile(t -> true)
                .verifyComplete();

    System.out.println("Total number of values in generated lists: " + adder.get());
}

将expectNextCount(1) 修改为expectNextCount(100_000)，我运行了一个打印Total number of values in generated lists: 102405 并花费了40 毫秒的运行。

Answer 2

如果您错过了reference guide，也许reference guide 可以让您走上正确的道路？

除了最常见的expectNext（您必须对序列中的每个项目重复）之外，如果您知道元素的数量，您可以使用expectNextCount，或者使用thenConsumeWhile基于谓词跳过元素，