サーバー WebSocketHandler は、SimpleUrlHandlerMapping および WebSocketHandlerAdapter を使用するリクエストにマップされます。クライアント WebSocketHandler が WebSocketClient 実行メソッドに渡されます。

session.receive() を使用して受信メッセージストリームを作成し、session.send(publisher) を使用して送信メッセージストリームを作成します。以下は、受信を処理し、送信メッセージを送信するための組み合わせフローの例です。

 class ExampleHandler implements WebSocketHandler {

        @Override
        public Mono<Void> handle(WebSocketSession session) {

                Flux<WebSocketMessage> output = session.receive()
                        .doOnNext(message -> {
                                // ...
                        })
                        .concatMap(message -> {
                                // ...
                        })
                        .map(value -> session.textMessage("Echo " + value));

                return session.send(output);
        }
 }
 

受信メッセージの処理と送信メッセージの送信が独立したストリームである場合、"zip" 演算子で結合できます。

 class ExampleHandler implements WebSocketHandler {

        @Override
        public Mono<Void> handle(WebSocketSession session) {

                Mono<Void> input = session.receive()
                        .doOnNext(message -> {
                                // ...
                        })
                        .concatMap(message -> {
                                // ...
                        })
                        .then();

                Flux<String> source = ... ;
                Mono<Void> output = session.send(source.map(session::textMessage));

                return Mono.zip(input, output).then();
        }
 }
 

WebSocketHandler は、受信ストリームと送信ストリームを統合フローに構成し、そのフローの完了を反映する Mono<Void> を返す必要があります。つまり、Reactive Streams シグナルはアクティビティを終了するため、接続が開いているかどうかを確認する必要はありません。受信ストリームは完了 / エラーシグナルを受信し、送信ストリームはキャンセルシグナルを受信します。