spring-websocket - 如何使用 Spring Reactive WebSocket 并将其转换为 Flux 流？

Question

有一些WebSocketClient关于 Spring documentation 的示例:

WebSocketClient client = new ReactorNettyWebSocketClient();
client.execute("ws://localhost:8080/echo"), session -> {...}).blockMillis(5000);

我不确定如何处理传入的数据流？
在那块里面 {...} .

我的意思是:如何过滤传入的数据并将其转换为 Flux？

这是我想要的。

@GetMapping("/stream", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<MyRecourse> getStreaming() {

    //  get some data from WebSocket (CoinCap service).
    //  Transform that data into MyRecourse object
    //  Return stream to a client 

}

Answer 1

看看那个 WebSocketSession WebSocketHandler.handle() 的参数 lambda :

/**
 * Get the flux of incoming messages.
 */
Flux<WebSocketMessage> receive();

见 Spring WebFlux Workshop想要查询更多的信息。

更新

让我们试试这个!

    Mono<Void> sessionMono =
            client.execute(new URI("ws://localhost:8080/echo"),
                    session ->
                            Mono.empty()
                                    .subscriberContext(Context.of(WebSocketSession.class, session))
                                    .then());

    return sessionMono
            .thenMany(
                    Mono.subscriberContext()
                            .flatMapMany(c -> c
                                    .get(WebSocketSession.class)
                                    .receive()))
            .map(WebSocketMessage::getPayloadAsText);

更新 2

或另一种选择，但订阅已被阻止:

    EmitterProcessor<String> output = EmitterProcessor.create();

    client.execute(new URI("ws://localhost:8080/echo"),
            session ->
                    session.receive()
                            .map(WebSocketMessage::getPayloadAsText)
                            .subscribeWith(output)
                            .then())
            .block(Duration.ofMillis(5000));

    return output;

更新 3

关于此事的有效 Spring Boot 应用程序: https://github.com/artembilan/webflux-websocket-demo

主要代码如下:

    EmitterProcessor<String> output = EmitterProcessor.create();

    Mono<Void> sessionMono =
            client.execute(new URI("ws://localhost:8080/echo"),
                    session -> session.receive()
                            .map(WebSocketMessage::getPayloadAsText)
                            .subscribeWith(output)
                            .then());

    return output.doOnSubscribe(s -> sessionMono.subscribe());