2种方式使用Spring webFlow流式传输

Question

我想知道是否有可能实现2种方式的流式传输使用Spring WebFlow？基本上，我希望让客户端发送服务器接收到的数据通量将它们映射到String，然后返回结果，所有这些都流利地进行，而无需收集数据。我使用RSocket完成了它，但我想知道我是否可以使用超文本传输协议2.0（使用Spring和Project-Retor）获得相同的结果。

试过这样做：

1-客户：

  public Mono<Void> stream() {
    var input = Flux.range(1, 10).delayElements(Duration.ofMillis(500));
    return stockWebClient.post()
            .uri("/stream")
            .body(BodyInserters.fromPublisher(input, Integer.class))
            .accept(MediaType.TEXT_EVENT_STREAM)
            .retrieve()
            .bodyToFlux(String.class)
            .log()
            .then();
  }

2.服务器：

@PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
  public Flux<String> stream(@RequestBody Integer i) {
    return Flux.range(i, i+10).map(n -> String.valueOf(i)).log();
  }

或者：

public Flux<String> stream(@RequestBody Flux<Integer> i) {
    return i.map(n -> String.valueOf(i)).log();
  }

或者：

@PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
  public Flux<String> stream(@RequestBody List<Integer> i) {
    return Flux.fromIterable(i).map(n -> String.valueOf(i)).log();
  }

没有一个工作正常。

Answer 1

如果你想使用服务器发送的事件，你需要返回一个Flux

所以你的服务器值应该是：

    @PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> stream(@RequestBody Integer i) {
        return Flux.range(i, i + 10).map(n -> ServerSentEvent.builder(String.valueOf(n)).build());
    }

但在这种情况下，主体只是一个整数，您的客户端代码变为：

input.flatMap(i ->
        stockWebClient
                .post()
                .uri("/stream")
                .bodyValue(i)
                .accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve()
                .bodyToFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
                .mapNotNull(ServerSentEvent::data)
                .log())
                .blockLast();

您也可以对功能endpoint执行相同的操作。

如果您希望能够将数据从客户端流到服务器，然后再流到服务器，那么您将无法使用SSE，但您可以通过websocket实现这一点。

你需要一个HandlerMapping和一个WebSocketHandler

public class TestWebSocketHandler implements WebSocketHandler {
    @Override
    public Mono<Void> handle(WebSocketSession session) {
        Flux<WebSocketMessage> output = session.receive()
                .map(WebSocketMessage::getPayloadAsText)
                .map(Integer::parseInt)
                .concatMap(i -> Flux.range(i, i + 10).map(String::valueOf))
                .map(session::textMessage);
        return session.send(output);
    }
}

处理程序的配置：

@Bean
    public TestWebSocketHandler myHandler() {
        return new TestWebSocketHandler();
    }

    @Bean
    public HandlerMapping handlerMapping(final TestWebSocketHandler myHandler) {
        Map<String, WebSocketHandler> map = new HashMap<>();
        map.put("/streamSocket", myHandler);
        int order = -1; // before annotated controllers
        return new SimpleUrlHandlerMapping(map, order);
    }

在客户端：

var input2 = Flux.range(1, 10).delayElements(Duration.ofMillis(500));
        WebSocketClient client = new ReactorNettyWebSocketClient();
        client.execute(URI.create("http://localhost:8080/streamSocket"), session ->
                session.send(input2.map(i -> session.textMessage("" + i))).then(session.receive().map(WebSocketMessage::getPayloadAsText).log().then())
        ).block();