Spring 5 Reactive WebSockets: recommended use

Question

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
• ...

Answer 1

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/