Reactive WebSockets with Spring

1. Overview

In this article, we’re going to create a quick example using the new Spring 5 WebSockets API along with reactive features provided by Spring WebFlux.

WebSocket is a well-known protocol that enables full-duplex communication between client and server, generally used in web applications where the client and server need to exchange events at high frequency and with low latency.

Spring Framework 5 has modernized WebSockets support in the framework, adding reactive capabilities to this communication channel.

We can find more on Spring WebFlux here.

2. Maven Dependencies

We’re going to use the spring-boot-starter dependencies for spring-boot-starter-webflux:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
    <version>3.3.0</version>
</dependency>

3. WebSocket Configuration in Spring

Our configuration is pretty straightforward: We’ll inject the WebSocketHandler to handle the socket session in our Spring WebSocket application.

@Autowired
private WebSocketHandler webSocketHandler;

Furthermore, let’s create a HandlerMapping bean-annotated method that will be responsible for the mapping between requests and handler objects:

@Bean
public HandlerMapping webSocketHandlerMapping() {
    Map<String, WebSocketHandler> map = new HashMap<>();
    map.put("/event-emitter", webSocketHandler);

    SimpleUrlHandlerMapping handlerMapping = new SimpleUrlHandlerMapping();
    handlerMapping.setOrder(1);
    handlerMapping.setUrlMap(map);
    return handlerMapping;
}

The URL we can connect to will be ws://localhost:<port>/event-emitter.

4. WebSocket Message Handling in Spring

Our ReactiveWebSocketHandler class will be responsible for managing the WebSocket session on the server side.

It implements the WebSocketHandler interface so we can override the handle method, which will be used to send the message to the WebSocket client:

@Component
public class ReactiveWebSocketHandler implements WebSocketHandler {
    
    // private fields ...

    @Override
    public Mono<Void> handle(WebSocketSession webSocketSession) {
        return webSocketSession.send(intervalFlux
          .map(webSocketSession::textMessage))
          .and(webSocketSession.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .log());
    }
}

5. Creating a Simple Reactive WebSocket Client

Let’s now create a Spring Reactive WebSocket client that will be able to connect and exchange information with our WebSocket server.

5.1. Maven Dependency

First, the Maven dependencies.

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

Here, we’re using the same spring-boot-starter-webflux used previously to set up our reactive WebSocket server application.

5.2. WebSocket Client

Now, let’s create the ReactiveClientWebSocket class, which is responsible for starting the communication with the server:

public class ReactiveJavaClientWebSocket {
 
    public static void main(String[] args) throws InterruptedException {
 
        WebSocketClient client = new ReactorNettyWebSocketClient();
        client.execute(
          URI.create("ws://localhost:8080/event-emitter"), 
          session -> session.send(
            Mono.just(session.textMessage("event-spring-reactive-client-websocket")))
            .thenMany(session.receive()
              .map(WebSocketMessage::getPayloadAsText)
              .log())
            .then())
            .block(Duration.ofSeconds(10L));
    }
}

In the code above, we can see that we’re using the ReactorNettyWebSocketClient, which is the WebSocketClient implementation for use with Reactor Netty.

Additionally, the client connects to the WebSocket server through the URL ws://localhost:8080/event-emitter, establishing a session as soon as it is connected to the server.

We can also see that we are sending a message to the server (“event-spring-reactive-client-websocket“) along with the connection request.

Furthermore, the method send is invoked, expecting as a parameter a variable of type Publisher<T>, which in our case our Publisher<T> is Mono<T> and T is a simple String “event-me-from-reactive-java-client-websocket“.

Moreover, the thenMany(…) method expecting a Flux of type String is invoked. The receive() method gets the flux of incoming messages, which later are converted into strings.

Finally, the block() method forces the client to disconnect from the server after the given time (10 seconds in our example).

5.3. Starting the Client

To run it, make sure the Reactive WebSocket Server is up and running. Then, launch the ReactiveJavaClientWebSocket class, and we can see on the sysout log the events being emitted:

[reactor-http-nio-4] INFO reactor.Flux.Map.1 - 
onNext({"eventId":"6042b94f-fd02-47a1-911d-dacf97f12ba6",
"eventDt":"2018-01-11T23:29:26.900"})

We also can see in the log from our Reactive WebSocket server the message sent by the client during the connection attempt:

[reactor-http-nio-2] reactor.Flux.Map.1: 
onNext(event-me-from-reactive-java-client)

Also, we can see the message of terminated connection after the client finished its requests (in our case, after the 10 seconds):

[reactor-http-nio-2] reactor.Flux.Map.1: onComplete()

6. Creating a Browser WebSocket Client

Let’s create a simple HTML/Javascript client, WebSocket, to consume our reactive WebSocket server application.

<div class="events"></div>
<script>
    var clientWebSocket = new WebSocket("ws://localhost:8080/event-emitter");
    clientWebSocket.onopen = function() {
        console.log("clientWebSocket.onopen", clientWebSocket);
        console.log("clientWebSocket.readyState", "websocketstatus");
        clientWebSocket.send("event-me-from-browser");
    }
    clientWebSocket.onclose = function(error) {
        console.log("clientWebSocket.onclose", clientWebSocket, error);
        events("Closing connection");
    }
    clientWebSocket.onerror = function(error) {
        console.log("clientWebSocket.onerror", clientWebSocket, error);
        events("An error occured");
    }
    clientWebSocket.onmessage = function(error) {
        console.log("clientWebSocket.onmessage", clientWebSocket, error);
        events(error.data);
    }
    function events(responseEvent) {
        document.querySelector(".events").innerHTML += responseEvent + "<br>";
    }
</script>

With the WebSocket server running, opening this HTML file in a browser (e.g., Chrome, Internet Explorer, Mozilla Firefox, etc.), we should see the events being printed on the screen, with a delay of 1 second per event, as defined in our WebSocket server.

{"eventId":"c25975de-6775-4b0b-b974-b396847878e6","eventDt":"2018-01-11T23:56:09.780"}
{"eventId":"ac74170b-1f71-49d3-8737-b3f9a8a352f9","eventDt":"2018-01-11T23:56:09.781"}
{"eventId":"40d8f305-f252-4c14-86d7-ed134d3e10c6","eventDt":"2018-01-11T23:56:09.782"}

7. Conclusion

Here, we’ve presented an example of creating a WebSocket communication between server and client using Spring 5 Framework, implementing the new reactive features provided by Spring Webflux.