如何使用Spring反应WebSocket并将其转换为通量流?

看看吧 WebSocketSession 参数 WebSocketHandler.handle() λ:

/**
 * 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());