在 Spring Boot Webflux 中生成服务器发送事件

Question

我在 SpringBoot 应用程序中有以下代码：

class MySse {
    public Mono<ServerResponse> emitEvents(ServerRequest request){
        return ServerResponse.ok()
                .contentType(MediaType.TEXT_EVENT_STREAM)
                .body(Mono.from(Flux.interval(Duration.ofSeconds(1))
                .map(sequence  -> ServerSentEvent.builder()
                        .id(String.valueOf(sequence))
                        .event("periodic-event")
                        .data("SSE - " + LocalTime.now().toString())
                        .build())), ServerSentEvent.class);
    }
}

@Configuration
public class MyRoutesConfiguration {

    @Autowired
    @Bean
    RouterFunction<ServerResponse> sseRoute(MySse mySse) {
        return route(path("/sse").and(method(HttpMethod.GET)), MySse ::emitEvents)
                ;
    }

    @Bean
    public MySse mySse() {
        return new MySse();
    }
}

如果我导航到 http://localhost（上面没有显示路由，但它可以工作） 从那里我在 Chrome 中打开 DevTools 并输入以下 JavaScript 代码：

const evtSource = new EventSource("sse/");
evtSource.onmessage = function(event) {
  console.log(event.data);
}

但是什么都没有打印出来……

MySse::emitEvent 中的 map(...) lambda 中的断点每秒被命中

但是在浏览器的 JS 控制台中什么都没有打印出来。

如果我访问 http://localhost/sse 我会得到以下响应：

id:0
event:periodic-event
data:SSE - 20:17:12.972706400

Answer 1

public Mono<ServerResponse> emitEvents(ServerRequest request){

    return ServerResponse.ok()
            .contentType(MediaType.TEXT_EVENT_STREAM)
            .body(BodyInserters.fromServerSentEvents(Flux.interval(Duration.ofSeconds(1))
            .map(aLong -> ServerSentEvent.<String>builder()
                            .id(String.valueOf(aLong))
                            .event("periodic-event")
                            .data("SSE - " + LocalTime.now().toString())
                            .build())));
}

如果您想流式传输数据，您需要返回一个Flux。 Mono 是 ONE 项，Flux 是 0...n 项。您触发Flux 以每隔一秒间隔开始发送，只要连接打开，它就会每秒向调用客户端发送一个事件。

您可以尝试使用 curl（或您正在使用的 chrome 控制台），但如果您使用 curl，则需要使用 -N 标志禁用 curl 缓冲策略。

curl -N http://localhost:8080/sse