Spring Reactive: java.io.IOException:在尝试单元测试返回流量的服务时，连接过早关闭

Question

在使用spring reactive应用程序时，我创建了一个每秒生成一个事件的rest服务。我的rest控制器的代码是：

@GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<Event> getEvents() {
    Flux<Event> eventFlux = Flux.fromStream(Stream.generate(() -> new Event(new Random().nextLong(), "Hello Event")));
    Flux<Long> emmitFlux = Flux.interval(Duration.ofSeconds(1));
    return Flux.zip(eventFlux, emmitFlux).map(Tuple2::getT1);
}

对此进行单元测试的方法如下：

webTestClient.get()
    .uri("/events")
    .accept(MediaType.TEXT_EVENT_STREAM)
    .exchange()
    .expectStatus()
    .isOk();

FluxExchangeResult<Event> result = webTestClient.get().uri("/events").accept(MediaType.TEXT_EVENT_STREAM)
    .exchange()
    .expectStatus()
    .isOk()
    .returnResult(Event.class);

Flux<Event> eventFlux = result.getResponseBody();

StepVerifier.create(eventFlux)        
    .expectSubscription()
    .thenAwait(Duration.ofSeconds(1))
    .expectNextCount(0)
    .thenAwait(Duration.ofSeconds(1))
    .expectNextCount(1)
    .thenAwait(Duration.ofSeconds(1))
    .expectNextCount(2);  

但是当我运行测试时，我得到了这个错误：

java.io.IOException: Connection closed prematurely

有没有人用spring-reactive解决过类似的问题？

Answer 1

您必须始终以.verify();结束StepVerifier链，否则它不会订阅它，也不会发生任何事情。

在这种情况下，因为它是无限流，所以在验证之前还必须有一个thenCancel()，否则测试可能会无限期地运行。