在Spring集成中使用Webflux或Flux生成端点消费服务器发送的事件

Question

如何通过Spring Integration使用服务器发送的事件？我知道Spring通过Webflux支持SSE，但是如何将传入的Flux转换为单独的消息实例？并可能将这些代码封装到某个Spring-Integration-Lifecycle感知组件中(MessageProducerSupport?)

WebClient client = WebClient.create("http://myhost:8080/sse");
ParameterizedTypeReference<ServerSentEvent<String>> type
 = new ParameterizedTypeReference<ServerSentEvent<String>>() {};

Flux<ServerSentEvent<String>> eventStream = client.get()
  .uri("/stream-sse")
  .retrieve()
  .bodyToFlux(type);

eventStream.subscribe(
  content -> ;/* here I believe the message should be produced/sent to a channel */ );

Answer 1

请参阅Spring Integration WebFlux出站网关：https://docs.spring.io/spring-integration/docs/current/reference/html/webflux.html#webflux-outbound

setExpectedResponseType(Class<?>)或setExpectedResponseTypeExpression(Expression)标识响应正文元素转换的目标类型。如果replyPayloadToFlux设置为true，则响应正文将转换为具有为每个元素提供的expectedResponseType的Flux，并将此Flux作为有效负载发送到下游。然后，您可以使用splitter以响应式的方式迭代此Flux。

WebFlux.outboundGateway("http://myhost:8080/sse/stream-sse")
            .httpMethod(HttpMethod.GET)
            .replyPayloadToFlux(true)
            .setExpectedResponseTypeExpression(new ParameterizedTypeReference<ServerSentEvent<String>>() {})

为了让它在应用程序准备就绪后立即开始工作，您可以实现一个ApplicationRunner，将“空”消息发送到具有该WebFlux.outboundGateway()的流的通道中。我不认为我们需要一个特殊的、专门的组件来处理SSE请求和生成。现有组件的组合就足够了。