服务器发送事件 Spring Webflux 返回事件名称以响应来自休息控制器的响应

Posted 2023-02-27

【中文标题】服务器发送事件 Spring Webflux 返回事件名称以响应来自休息控制器的响应

【英文标题】：Server Sent Event Spring Webflux return event name in response from rest controller

【发布时间】：2020-07-14 12:33:39

【问题描述】：

我正在使用 SSE 和 reactor Flux 来捕获 linux 系统指标并发布到浏览器（javascript 事件源）。

目前，问题是无法将特定数据发送到特定客户端。在浏览了多个 SO 帖子后，它被告知在事件源中使用事件侦听器，并且来自服务器的响应应该将 event 作为响应中的键。

既然我从服务器返回一个对象，那么将 event 作为响应 json 中的键之一是否足够，它将由事件源识别。

在客户端设置一个唯一编号，将在json响应中的event键中返回。

感谢您的帮助！

Javascript：

jsonStreamObjectHeap.addEventListener("197e08e-f9a4-4e6e-9a04-220ade08a8f4",function(e)
        $.each(message, function(index, value) 
           /*some operation*/
        );
        

来自 Rest Controller 的响应：

   "event":"197e08e-f9a4-4e6e-9a04-220ade08a8f4",
   "data":"2048"

休息控制器代码：

 Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));
        interval.subscribe((i) -> testStreamList.forEach(testStream -> 
            try 
                generateTestStream(testStream, UUID);
             catch (JsonProcessingException e) 
                e.printStackTrace();
            
        ));
 Flux<List<TestStream>> transactionFlux = Flux.fromStream(Stream.generate(() -> testStreamList));
 return Flux.zip(interval, transactionFlux).map(Tuple2::getT2);

【问题讨论】：

订阅者是调用客户端，你的应用是发布者，所以不要订阅发布者。所以删除“订阅”。其次，generateTestStream 和 testStramList 是什么。代码很难理解，我是否正确假设您要做的就是以给定的时间间隔（1 秒）从您的testStreamList 返回一个对象。这个对象需要是 SSE 吗？如果那是正确的，那么这是一种乱七八糟的写法，而且你写它的方式也是阻塞的。

@ThomasAndolf 你是对的。你能告诉我，我该如何继续？或任何链接/帮助？

【参考方案1】：

我发现 org.springframework.http.codec.ServerSentEvent 可以插入到来自其余控制器的 Flux 响应中。

  Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));    
  Flux<HeapStat> transactionFlux = Flux.fromStream(Stream.generate(() -> heapStat));

  /*Bulding ServerSentEvent to from the tuple*/

  return Flux.zip(interval, transactionFlux).map(tuple->
           ServerSentEvent.<HeapStat>builder().event(jsessionId).data(tuple.getT2()).build()
                );

您可以准备与 SSEEmmiter 完全一样的响应，带有自定义事件名称、Last-Event-ID

【讨论】：

在 Flux.zip 中发送错误而不是事件的任何方式？