问题

我正在尝试按照以下教程 https://docs.spring.io/spring/docs/current/spring-framework-reference/testing.html#webtestclient-stream 中的说明测试我的 SSE API。

不幸的是，对我来说这并不起作用。要生成下一个事件，需要执行一些将触发它的操作。在等待 WebTestClient 的响应时，我无法做到这一点（我没有找到添加这种处理程序的可能性）。

我通过创建一个单独的线程来找到了解决方法，以便定期触发生成事件，但这并不优雅。有没有更好的方法来做到这一点？

TimerTask task = new TimerTask() {
public void run() {
while (true) {
// 触发定期生成事件的代码
}
}
};
Timer timer = new Timer("Timer");
long delay = 1000L;
timer.schedule(task, delay);

FluxExchangeResult result = client.get().uri("/events")
.accept(TEXT_EVENT_STREAM)
.exchange()
.expectStatus().isOk()
.returnResult(MyEvent.class);

Flux eventFlux = result.getResponseBody();

StepVerifier.create(eventFlux)
.expectNext(person)
.expectNextCount(4)
.consumeNextWith(p -> ...)
.thenCancel()
.verify();

<details>
<summary>英文:</summary>

I am trying to test my SSE API in the way explained in the following tutorial https://docs.spring.io/spring/docs/current/spring-framework-reference/testing.html#webtestclient-stream .

Unfortunately, it doesn&#39;t work for me. To generate next event it is necessary to perform some actions that will trigger it. I couldn&#39;t do it while waiting for response from WebTestClient (I didn&#39;t find the possibility to add such handler).

I found the workaround by creating separate thread that trigger generating events periodically, but it is not elegant. Is there any better way to do it?

TimerTask task = new TimerTask() {
public void run() {
while (true) {
//code that trigger generating event periodically
}
}
};
Timer timer = new Timer("Timer");
long delay = 1000L;
timer.schedule(task, delay);

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

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

StepVerifier.create(eventFlux)
.expectNext(person)
.expectNextCount(4)
.consumeNextWith(p -> ...)
.thenCancel()
.verify();

</details>


# 答案1
**得分**: 0

我遇到过类似的问题。如果我理解你的意思正确，你想要发布某种消息，然后检查是否收到了该消息。如果是这种情况，你应该使用订阅并提供消费者。以下是一个示例代码：

```java
FluxExchangeResult<MyEvent> result = client.get().uri("/events")
    .accept(TEXT_EVENT_STREAM)
    .exchange()
    .expectStatus().isOk()
    .returnResult(MyEvent.class)
    .getResponseBody()
    .subscribe(new Consumer<MyEvent>() {
        @Override
        public void accept(MyEvent event) {
            // 在这里处理事件
        }
    });

FluxExchangeResult<MyEvent> result = client.get().uri("/events")
    .accept(TEXT_EVENT_STREAM)
    .exchange()
    .expectStatus().isOk()
    .returnResult(MyEvent.class)
    .getResponseBody()
    .subscribe(new Consumer<MyEvent>() {
        @Override
        public void accept(String event) {
        }
    });