Reactive and scalable chat with Kotlin + Spring + WebSockets

Content

1. Project configuration
2. Logger
3. Domain
4. Mapper
5. Spring Security configuration
6. WebSocket configuration
7. Solution architecture
8. Implementation
9. Integration with Redis
10. Implementation service layer
11. Conclusion

Introducing

This tutorial is about implementation reactive scalable chat with Kotlin and Spring using WebSocket as a communication approach.

Let’s get started

For the first, we need to configure our Logger.

Configuration of logger represents a prototype bean that will configure the logger for class retrieved from the injection point. This approach is above

Now to get a logger we just should inject wherever we want.

Create domain and configure object mapper for it

Class Chat represents base information, including chat members.

Class ChatMember is describing chat members. The purpose of the deleted flag is to remove it from chat selection for users with userId.

The code snippet above shows the base class for all chat messages. Annotation @JsonTypeInfo needs for Kotlin can determinate subtype for inherited classes.

The next code snippet including specific chat message class that represent text message in chat.

Let’s configure ObjectMapper

There is not only base configuration for dates, Kotlin modules and so, but also added configuration for subtypes of WebSocketEvents and subtypes of CommonMessages for example I added two subtypes.

Spring Security Configuration

For the start, we should create ReactiveAuthenticationManager and SecurityContextRepository. The authentication approach we use is JWT, so create class JwtAuthentication class with next code:

To extract information about authenticated users wherever we need put claims to security context details.

We will support two approaches to retrieve tokens from request:

1. Header Authorization: Bearer ${JWT_TOKEN}
2. GET param ?access_token=${JWT_TOKEN}

Now we have needed classes to finish our configuration up.

Here all requests to our application should be authenticated. But for all paths started with /ws/ we expect that there is a specific role: ROLE_USER.

We’ve finished with Security config, now we need to configure sockets.

WebSocket Configuration

First we need to set mapping between uri and handler.

1. We create a map, where key is URI, and the value is a handler.
2. Create a handler for the mapping.

Here for URI /ws/chat we set handler to ChatWebSocketHandler. ChatWebSocketHandler without implementation is above.

Solution architecture

Now we can’t scale our chat application. Because WebSockets has a duplex connection and establishes a connection with an instance we will send all our messages to it. So if we will start more than one instance we will haven’t access to all existing sessions. This way we have a case when chat members connected to different instances and there is no way to communicate between them directly. To solve this problem we will use message broker that will get a message and broadcast this message to all instances. Instances will be looking for members of the chat and if there is send them the message.

Implementation

Let’s implement our handler ChatWebSocketHandler.

First, create a map where we will map userId to socket session. To handle cases when one user connected throw a few devices we have the next interface of the map: MutableMap<UUID, LinkedList<WebSocketSession>>

Create an entry in the map we will be when getting a subscriber to session.receive stream, and removing when the subscriber unsubscribes.

In method getReceiverStream we create stream-handler for incoming messages from a client. We get payload as a string and convert it to WebSocketEvent, after that depending on the type pass it to the service layer

In method getSenderStream we configure a stream that sending messages to the client by sockets.

For writing to the sockets we should create a stream and create methods to write to this stream data. Since reactor 3.4 for this purpose Sinks.Many use recommended. Create a stream in SinkWrapper class.

Now If we will send data to this stream it will be handled in the stream that was created in getSenderStream.

Integration with Redis

Redis has PUB/SUB model communication, which one perfectly use for our solution.

So we need:

1. RedisChatMessageListener — subscription for topics and redirecting incoming messages to the service layer.
2. RedisChatMessagePublisher — publishing messages to topics
3. RedisConfig — configuration for Redis
4. RedisListenerStarter — starting our listeners

Implementation:

RedisConfig is common, nothing interesting

RedisChatMessageListener

Here we create a subscription to the topic by base class name(usually topic names extract to properties). When a message is retrieved it will be converted to an object and passed to sendMessage, which will determine chat members and will try to send them the message.

RedisChatMessagePublisher

The publisher has one method to broadcast an object of CommonMessage type to all instances. The object will be converted to the string and published to the topic by base class name.

RedisListenerStarter

In this class starts all listeners from RedisChatMessageListener. In our case — there is only one listener subscribeOnCommonMessageTopic.

Service implementation

Base implementation, without saving to DB and mocking chatRepository.

Method handleNewMessageEvent invoking by WebSocketHandler and get userId of sender and NewMessageEvent — common text message. This method will check wherever or not the sender is a member of this chat and after that the message broadcasting between instances.

Conclusion

In case to improve the application you can separate receiving and sending events to different classes. Also the method of receiving of WebSocketEvents and passing it to handler can be improved with removing hardocde of mapping type => handler.

Project on GitHub: https://github.com/bogatikov/reactive-chat