Spring 5 Reactive WebSockets: recommended use

user_775138 0
Viewed 2594

I've been learning a bit about Spring 5 WebFlux, reactive programming and websockets. I've watched Josh Long's Spring Tips: Reactive WebSockets with Spring Framework 5. The code that sends data from server to client through a WebSocket connection uses a Spring Integration IntegrationFlow that publishes to a PublishSubcribeChannel which has a custom MessageHandler subscribed to it that takes the message, converts it to an object that is then converted to Json and emitted to the FluxSink from the callback supplied to Flux.create(), which is used to send to the WebSocketConnection.

I was wondering if the use of IntegrationFlow and PublishSubscribeChannel is the recommended way to push events from a background process to the client, or if this is just more convenient in this particular example (monitoring the file system). I'd think if you have control over the background process, you could have it emit to the FluxSink directly?

I'm thinking about use cases similar to the following:

  • a machine learning process whose progress is monitored
  • updates to the state of a game world that are sent to players
  • chat rooms / team collaboration software
  • ...
spring-webflux reactive-programming spring websocket
2 Answers

What I've done in the past that has worked for me is to create a Spring Component that implements WebSocketHandler:

@Component
public class ReactiveWebSocketHandler implements WebSocketHandler {

Then in the handle method, Spring injects the WebSocketSession object

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

Then create one or more Flux reactive publishers that emit messages(WebSocketMessage) for the client.

    final var output = session.send(Flux.merge(flux1, flux2));

Then you can zip up the incoming and outgoing Flux objects in a Mono and then Spring will take it from there.

        return Mono.zip(incomingWebsocketMsgResponse.getWebSocketMsgFlux().then(),
                outputWithErrorMsgs)
                .then();

Example: https://howtodoinjava.com/spring-webflux/reactive-websockets/

user_5551770

Since this question, Spring introduced RSocket support - you might think about it like the WebSocket STOMP support existing in Spring MVC, but much more powerful and efficient, supporting backpressure and advanced communication patterns at the protocol level.

For the use cases you're mentioning, I'd advise using RSocket as you'd get a powerful programming model with @MessageMapping and all the expected support in Spring (codecs for JSON and CBOR, security, etc).

user_229820
Related