Mahi Mullapudi

Posted on Jul 15

Building Real-Time Applications with WebSockets and Reactive Streams

#java #springboot #reactiveprogramming #websockets

Introduction

Real-time applications are becoming increasingly important in today's digital world, providing users with instant updates and interactive experiences. Technologies like WebSockets and Reactive Streams play a crucial role in enabling these real-time capabilities. In this article, we'll explore how to build a real-time chat application using WebSockets and Reactive Streams with Java and Spring Boot. We'll cover all the necessary concepts and provide detailed examples to help you get started.

Understanding WebSockets

What are WebSockets?

WebSockets are a communication protocol that provides full-duplex communication channels over a single TCP connection. Unlike traditional HTTP, which follows a request-response model, WebSockets allow for continuous two-way interaction between the client and server. This makes WebSockets ideal for applications that require real-time data updates, such as chat applications, live notifications, and online gaming.

How WebSockets Work

Handshake: The WebSocket connection starts with a handshake request from the client to the server. This request is an HTTP request upgraded to a WebSocket connection.
Connection Establishment: If the server accepts the handshake request, the connection is established, and both parties can start sending and receiving messages.
Data Transfer: Once the connection is established, messages can be sent in both directions at any time. Messages are transmitted in frames, which can be either text or binary data.
Connection Closure: Either the client or server can close the connection by sending a close frame. The other party responds with a close frame to complete the closure.

Benefits of WebSockets

Low Latency: WebSockets provide low-latency communication, making them suitable for real-time applications.
Persistent Connection: Once established, the WebSocket connection remains open, reducing the overhead of establishing multiple HTTP connections.
Bidirectional Communication: Both the client and server can send messages independently, enabling real-time updates.

WebSocket Lifecycle

Connection Establishment: The client sends a handshake request to the server to establish a WebSocket connection.
Data Transfer: Once the connection is established, the client and server can exchange messages in both directions.
Connection Closure: Either the client or the server can close the connection when it's no longer needed.

Introduction to Reactive Streams

What are Reactive Streams?

Reactive Streams is a specification for asynchronous stream processing with non-blocking backpressure. It provides a standard for handling asynchronous data streams, allowing you to build resilient and responsive applications. Reactive Streams are particularly useful in environments where you need to handle a large number of concurrent data streams efficiently.

Advantages of Reactive Programming

Asynchronous Processing: Reactive Streams allow you to handle data asynchronously, improving the scalability and performance of your application.
Non-blocking Backpressure: Reactive Streams provide mechanisms to handle backpressure, ensuring that your application can handle varying data rates without overwhelming the system.
Composability: Reactive Streams offer a composable approach to building complex data pipelines, making it easier to manage and reason about your code.

Key Concepts in Reactive Streams

Publisher: Produces data and sends it to Subscribers.
Subscriber: Consumes data produced by Publishers.
Subscription: Represents the relationship between a Publisher and a Subscriber.
Processor: Acts as both a Publisher and a Subscriber, enabling data transformation and processing.

Combining WebSockets with Reactive Streams

The Synergy Between WebSockets and Reactive Streams

Combining WebSockets with Reactive Streams allows you to build powerful real-time applications. WebSockets provide the communication channel, while Reactive Streams handle the data flow and processing. This combination enables you to build applications that are both responsive and scalable.

Architecture Overview

WebSocket Connection: Establish a WebSocket connection between the client and server.
Reactive Stream Processing: Use Reactive Streams to handle incoming and outgoing data.
Data Broadcasting: Broadcast data to multiple clients using WebSockets.

Use Case: Live Chat Application

Why Choose a Chat Application?

A live chat application is an excellent example to demonstrate the capabilities of WebSockets and Reactive Streams. It requires real-time communication, scalability to handle multiple users, and efficient data processing. This makes it an ideal use case to showcase how these technologies work together.

Features of the Chat Application

Real-time messaging between users.
Broadcast messages to all connected clients.
Handle multiple users simultaneously.
Efficiently manage data streams and backpressure.

Implementation with Java and Spring Boot

Setting Up the Project

Create a Spring Boot Project: Use Spring Initializr to create a new Spring Boot project with WebSocket and Reactive Web dependencies. ```shell

spring init --dependencies=websocket,reactive-web chat-application


2. **Project Structure**:

├── src
│ ├── main
│ │ ├── java
│ │ │ └── com.example.chat
│ │ │ ├── ChatApplication.java
│ │ │ ├── config
│ │ │ │ └── WebSocketConfig.java
│ │ │ ├── controller
│ │ │ │ └── ChatController.java
│ │ │ ├── handler
│ │ │ │ └── ChatWebSocketHandler.java
│ │ │ ├── model
│ │ │ │ └── ChatMessage.java
│ │ │ └── service
│ │ │ └── ChatService.java
│ │ ├── resources
│ │ │ └── application.properties



  
  
  WebSocket Configuration


Configure WebSockets in Spring Boot by creating a WebSocket configuration class.



package com.example.chat.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
import com.example.chat.handler.ChatWebSocketHandler;

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    private final ChatWebSocketHandler chatWebSocketHandler;

    public WebSocketConfig(ChatWebSocketHandler chatWebSocketHandler) {
        this.chatWebSocketHandler = chatWebSocketHandler;
    }

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(chatWebSocketHandler, "/chat").setAllowedOrigins("*");
    }
}


</code></pre></div><h3>
  <a name="explanation-of-keywords" href="#explanation-of-keywords">
  </a>
  Explanation of Keywords
</h3>

<ul>
<li>
<strong>@Configuration</strong>: Indicates that the class declares one or more @Bean methods and may be processed by the Spring container to generate bean definitions and service requests at runtime.</li>
<li>
<strong>@EnableWebSocket</strong>: Enables WebSocket support in a Spring application.</li>
<li>
<strong>WebSocketConfigurer</strong>: An interface to configure WebSocket handlers.</li>
<li>
<strong>WebSocketHandlerRegistry</strong>: A registry for WebSocket handlers.</li>
<li>
<strong>setAllowedOrigins</strong>: Configures allowed origins for cross-origin requests.</li>
</ul>
<h3>
  <a name="creating-the-websocket-handler" href="#creating-the-websocket-handler">
  </a>
  Creating the WebSocket Handler
</h3>

<p>Implement the WebSocket handler to manage chat messages.</p>
<div class="highlight"><pre class="highlight java"><code>

<span class="kn">package</span> <span class="nn">com.example.chat.handler</span><span class="o">;</span>

<span class="kn">import</span> <span class="nn">com.example.chat.model.ChatMessage</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">com.example.chat.service.ChatService</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">com.fasterxml.jackson.databind.ObjectMapper</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.web.socket.TextMessage</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.web.socket.handler.TextWebSocketHandler</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.web.socket.WebSocketSession</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.stereotype.Component</span><span class="o">;</span>

<span class="nd">@Component</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">ChatWebSocketHandler</span> <span class="kd">extends</span> <span class="nc">TextWebSocketHandler</span> <span class="o">{</span>

    <span class="kd">private</span> <span class="kd">final</span> <span class="nc">ChatService</span> <span class="n">chatService</span><span class="o">;</span>
    <span class="kd">private</span> <span class="kd">final</span> <span class="nc">ObjectMapper</span> <span class="n">objectMapper</span><span class="o">;</span>

    <span class="kd">public</span> <span class="nf">ChatWebSocketHandler</span><span class="o">(</span><span class="nc">ChatService</span> <span class="n">chatService</span><span class="o">,</span> <span class="nc">ObjectMapper</span> <span class="n">objectMapper</span><span class="o">)</span> <span class="o">{</span>
        <span class="k">this</span><span class="o">.</span><span class="na">chatService</span> <span class="o">=</span> <span class="n">chatService</span><span class="o">;</span>
        <span class="k">this</span><span class="o">.</span><span class="na">objectMapper</span> <span class="o">=</span> <span class="n">objectMapper</span><span class="o">;</span>
    <span class="o">}</span>

    <span class="nd">@Override</span>
    <span class="kd">public</span> <span class="kt">void</span> <span class="nf">handleTextMessage</span><span class="o">(</span><span class="nc">WebSocketSession</span> <span class="n">session</span><span class="o">,</span> <span class="nc">TextMessage</span> <span class="n">message</span><span class="o">)</span> <span class="kd">throws</span> <span class="nc">Exception</span> <span class="o">{</span>
        <span class="nc">String</span> <span class="n">payload</span> <span class="o">=</span> <span class="n">message</span><span class="o">.</span><span class="na">getPayload</span><span class="o">();</span>
        <span class="nc">ChatMessage</span> <span class="n">chatMessage</span> <span class="o">=</span> <span class="n">objectMapper</span><span class="o">.</span><span class="na">readValue</span><span class="o">(</span><span class="n">payload</span><span class="o">,</span> <span class="nc">ChatMessage</span><span class="o">.</span><span class="na">class</span><span class="o">);</span>
        <span class="n">chatService</span><span class="o">.</span><span class="na">sendMessage</span><span class="o">(</span><span class="n">chatMessage</span><span class="o">);</span>
    <span class="o">}</span>
<span class="o">}</span>


</code></pre></div><h3>
  <a name="explanation-of-keywords" href="#explanation-of-keywords">
  </a>
  Explanation of Keywords
</h3>

<ul>
<li>
<strong>@Component</strong>: Indicates that an annotated class is a "component". Such classes are considered as candidates for auto-detection when using annotation-based configuration and classpath scanning.</li>
<li>
<strong>TextWebSocketHandler</strong>: A convenience base class for handling WebSocket text messages.</li>
<li>
<strong>WebSocketSession</strong>: Represents a WebSocket session between a client and server.</li>
<li>
<strong>TextMessage</strong>: Represents a WebSocket text message.</li>
</ul>
<h3>
  <a name="reactive-streams-for-chat-messages" href="#reactive-streams-for-chat-messages">
  </a>
  Reactive Streams for Chat Messages
</h3>

<p>Implement a service to handle chat messages using Reactive Streams.</p>
<div class="highlight"><pre class="highlight java"><code>

<span class="kn">package</span> <span class="nn">com.example.chat.service</span><span class="o">;</span>

<span class="kn">import</span> <span class="nn">com.example.chat.model.ChatMessage</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.stereotype.Service</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">reactor.core.publisher.EmitterProcessor</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">reactor.core.publisher.Flux</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">reactor.core.publisher.FluxSink</span><span class="o">;</span>

<span class="nd">@Service</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">ChatService</span> <span class="o">{</span>

    <span class="kd">private</span> <span class="kd">final</span> <span class="nc">EmitterProcessor</span><span class="o">&lt;</span><span class="nc">ChatMessage</span><span class="o">&gt;</span> <span class="n">chatProcessor</span> <span class="o">=</span> <span class="nc">EmitterProcessor</span><span class="o">.</span><span class="na">create</span><span class="o">();</span>
    <span class="kd">private</span> <span class="kd">final</span> <span class="nc">FluxSink</span><span class="o">&lt;</span><span class="nc">ChatMessage</span><span class="o">&gt;</span> <span class="n">chatSink</span> <span class="o">=</span> <span class="n">chatProcessor</span><span class="o">.</span><span class="na">sink</span><span class="o">();</span>

    <span class="kd">public</span> <span class="nc">Flux</span><span class="o">&lt;</span><span class="nc">ChatMessage</span><span class="o">&gt;</span> <span class="nf">getChatMessages</span><span class="o">()</span> <span class="o">{</span>
        <span class="k">return</span> <span class="n">chatProcessor</span><span class="o">.</span><span class="na">publish</span><span class="o">().</span><span class="na">autoConnect</span><span class="o">();</span>
    <span class="o">}</span>

    <span class="kd">public</span> <span class="kt">void</span> <span class="nf">sendMessage</span><span class="o">(</span><span class="nc">ChatMessage</span> <span class="n">message</span><span class="o">)</span> <span class="o">{</span>
        <span class="n">chatSink</span><span class="o">.</span><span class="na">next</span><span class="o">(</span><span class="n">message</span><span class="o">);</span>
    <span class="o">}</span>
<span class="o">}</span>


</code></pre></div><h3>
  <a name="explanation-of-keywords" href="#explanation-of-keywords">
  </a>
  Explanation of Keywords
</h3>

<ul>
<li>
<strong>@Service</strong>: Indicates that an annotated class is a "Service". Such classes are considered as candidates for auto-detection when using annotation-based configuration and classpath scanning.</li>
<li>**EmitterProcessor</li>
</ul>

<p>**: A processor that allows dynamic push-pull flow control.</p>

<ul>
<li>
<strong>Flux</strong>: A Reactive Streams Publisher with RxJava 2.x API and backpressure support.</li>
<li>
<strong>FluxSink</strong>: An interface through which subscribers receive items.</li>
</ul>
<h3>
  <a name="websocket-controller" href="#websocket-controller">
  </a>
  WebSocket Controller
</h3>

<p>Create a WebSocket controller to manage chat message broadcasting.</p>
<div class="highlight"><pre class="highlight java"><code>

<span class="kn">package</span> <span class="nn">com.example.chat.controller</span><span class="o">;</span>

<span class="kn">import</span> <span class="nn">com.example.chat.model.ChatMessage</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">com.example.chat.service.ChatService</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.messaging.handler.annotation.MessageMapping</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.messaging.handler.annotation.SendTo</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.stereotype.Controller</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.springframework.web.bind.annotation.SubscribeMapping</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">reactor.core.publisher.Flux</span><span class="o">;</span>

<span class="nd">@Controller</span>
<span class="kd">public</span> <span class="kd">class</span> <span class="nc">ChatController</span> <span class="o">{</span>

    <span class="kd">private</span> <span class="kd">final</span> <span class="nc">ChatService</span> <span class="n">chatService</span><span class="o">;</span>

    <span class="kd">public</span> <span class="nf">ChatController</span><span class="o">(</span><span class="nc">ChatService</span> <span class="n">chatService</span><span class="o">)</span> <span class="o">{</span>
        <span class="k">this</span><span class="o">.</span><span class="na">chatService</span> <span class="o">=</span> <span class="n">chatService</span><span class="o">;</span>
    <span class="o">}</span>

    <span class="nd">@MessageMapping</span><span class="o">(</span><span class="s">"/chat.sendMessage"</span><span class="o">)</span>
    <span class="nd">@SendTo</span><span class="o">(</span><span class="s">"/topic/public"</span><span class="o">)</span>
    <span class="kd">public</span> <span class="nc">ChatMessage</span> <span class="nf">sendMessage</span><span class="o">(</span><span class="nc">ChatMessage</span> <span class="n">chatMessage</span><span class="o">)</span> <span class="o">{</span>
        <span class="n">chatService</span><span class="o">.</span><span class="na">sendMessage</span><span class="o">(</span><span class="n">chatMessage</span><span class="o">);</span>
        <span class="k">return</span> <span class="n">chatMessage</span><span class="o">;</span>
    <span class="o">}</span>

    <span class="nd">@SubscribeMapping</span><span class="o">(</span><span class="s">"/chat.getMessages"</span><span class="o">)</span>
    <span class="kd">public</span> <span class="nc">Flux</span><span class="o">&lt;</span><span class="nc">ChatMessage</span><span class="o">&gt;</span> <span class="nf">getMessages</span><span class="o">()</span> <span class="o">{</span>
        <span class="k">return</span> <span class="n">chatService</span><span class="o">.</span><span class="na">getChatMessages</span><span class="o">();</span>
    <span class="o">}</span>
<span class="o">}</span>


</code></pre></div><h3>
  <a name="explanation-of-keywords" href="#explanation-of-keywords">
  </a>
  Explanation of Keywords
</h3>

<ul>
<li>
<strong>@Controller</strong>: Indicates that an annotated class is a "Controller" (e.g., a web controller).</li>
<li>
<strong>@MessageMapping</strong>: Maps a message to a specific handler method.</li>
<li>
<strong>@SendTo</strong>: Specifies the destination to send the return value of a message-handling method.</li>
<li>
<strong>@SubscribeMapping</strong>: Maps a subscription to a specific handler method.</li>
</ul>
<h3>
  <a name="clientside-implementation" href="#clientside-implementation">
  </a>
  Client-Side Implementation
</h3>

<p>Implement the client-side using JavaScript with WebSocket.</p>
<div class="highlight"><pre class="highlight html"><code>

<span class="cp">&lt;!DOCTYPE html&gt;</span>
<span class="nt">&lt;html&gt;</span>
<span class="nt">&lt;head&gt;</span>
    <span class="nt">&lt;title&gt;</span>Chat Application<span class="nt">&lt;/title&gt;</span>
    <span class="nt">&lt;script&gt;</span>
        <span class="kd">let</span> <span class="nx">socket</span> <span class="o">=</span> <span class="k">new</span> <span class="nc">WebSocket</span><span class="p">(</span><span class="dl">"</span><span class="s2">ws://localhost:8080/chat</span><span class="dl">"</span><span class="p">);</span>

        <span class="nx">socket</span><span class="p">.</span><span class="nx">onmessage</span> <span class="o">=</span> <span class="kd">function</span><span class="p">(</span><span class="nx">event</span><span class="p">)</span> <span class="p">{</span>
            <span class="kd">let</span> <span class="nx">message</span> <span class="o">=</span> <span class="nx">JSON</span><span class="p">.</span><span class="nf">parse</span><span class="p">(</span><span class="nx">event</span><span class="p">.</span><span class="nx">data</span><span class="p">);</span>
            <span class="nf">displayMessage</span><span class="p">(</span><span class="nx">message</span><span class="p">);</span>
        <span class="p">};</span>

        <span class="kd">function</span> <span class="nf">sendMessage</span><span class="p">()</span> <span class="p">{</span>
            <span class="kd">let</span> <span class="nx">messageContent</span> <span class="o">=</span> <span class="nb">document</span><span class="p">.</span><span class="nf">getElementById</span><span class="p">(</span><span class="dl">"</span><span class="s2">message</span><span class="dl">"</span><span class="p">).</span><span class="nx">value</span><span class="p">;</span>
            <span class="kd">let</span> <span class="nx">message</span> <span class="o">=</span> <span class="p">{</span>
                <span class="na">sender</span><span class="p">:</span> <span class="dl">"</span><span class="s2">User</span><span class="dl">"</span><span class="p">,</span>
                <span class="na">content</span><span class="p">:</span> <span class="nx">messageContent</span>
            <span class="p">};</span>
            <span class="nx">socket</span><span class="p">.</span><span class="nf">send</span><span class="p">(</span><span class="nx">JSON</span><span class="p">.</span><span class="nf">stringify</span><span class="p">(</span><span class="nx">message</span><span class="p">));</span>
            <span class="nb">document</span><span class="p">.</span><span class="nf">getElementById</span><span class="p">(</span><span class="dl">"</span><span class="s2">message</span><span class="dl">"</span><span class="p">).</span><span class="nx">value</span> <span class="o">=</span> <span class="dl">''</span><span class="p">;</span>
        <span class="p">}</span>

        <span class="kd">function</span> <span class="nf">displayMessage</span><span class="p">(</span><span class="nx">message</span><span class="p">)</span> <span class="p">{</span>
            <span class="kd">let</span> <span class="nx">messageElement</span> <span class="o">=</span> <span class="nb">document</span><span class="p">.</span><span class="nf">createElement</span><span class="p">(</span><span class="dl">'</span><span class="s1">div</span><span class="dl">'</span><span class="p">);</span>
            <span class="nx">messageElement</span><span class="p">.</span><span class="nx">textContent</span> <span class="o">=</span> <span class="s2">`</span><span class="p">${</span><span class="nx">message</span><span class="p">.</span><span class="nx">sender</span><span class="p">}</span><span class="s2">: </span><span class="p">${</span><span class="nx">message</span><span class="p">.</span><span class="nx">content</span><span class="p">}</span><span class="s2">`</span><span class="p">;</span>
            <span class="nb">document</span><span class="p">.</span><span class="nf">getElementById</span><span class="p">(</span><span class="dl">'</span><span class="s1">messages</span><span class="dl">'</span><span class="p">).</span><span class="nf">appendChild</span><span class="p">(</span><span class="nx">messageElement</span><span class="p">);</span>
        <span class="p">}</span>
    <span class="nt">&lt;/script&gt;</span>
<span class="nt">&lt;/head&gt;</span>
<span class="nt">&lt;body&gt;</span>
    <span class="nt">&lt;h1&gt;</span>Chat Application<span class="nt">&lt;/h1&gt;</span>
    <span class="nt">&lt;div</span> <span class="na">id=</span><span class="s">"messages"</span><span class="nt">&gt;&lt;/div&gt;</span>
    <span class="nt">&lt;input</span> <span class="na">type=</span><span class="s">"text"</span> <span class="na">id=</span><span class="s">"message"</span> <span class="na">placeholder=</span><span class="s">"Enter your message"</span><span class="nt">&gt;</span>
    <span class="nt">&lt;button</span> <span class="na">onclick=</span><span class="s">"sendMessage()"</span><span class="nt">&gt;</span>Send<span class="nt">&lt;/button&gt;</span>
<span class="nt">&lt;/body&gt;</span>
<span class="nt">&lt;/html&gt;</span>


</code></pre></div><h3>
  <a name="explanation-of-keywords" href="#explanation-of-keywords">
  </a>
  Explanation of Keywords
</h3>

<ul>
<li>
<strong>WebSocket</strong>: Creates a new WebSocket connection to the specified URL.</li>
<li>
<strong>onmessage</strong>: An event handler that is called when a message is received from the server.</li>
<li>
<strong>send</strong>: Sends data to the server over the WebSocket connection.</li>
</ul>
<h3>
  <a name="testing-and-debugging" href="#testing-and-debugging">
  </a>
  Testing and Debugging
</h3>
<h4>
  <a name="testing-websocket-endpoints" href="#testing-websocket-endpoints">
  </a>
  Testing WebSocket Endpoints
</h4>

<ul>
<li>Use tools like Postman or WebSocket clients to test the WebSocket endpoints.</li>
<li>Write unit tests to validate the WebSocket configuration and handlers.</li>
</ul>
<h4>
  <a name="debugging-tips" href="#debugging-tips">
  </a>
  Debugging Tips
</h4>

<ul>
<li>Enable detailed logging for WebSocket events.</li>
<li>Use browser developer tools to monitor WebSocket connections and messages.</li>
</ul>
<h2>
  <a name="conclusion" href="#conclusion">
  </a>
  Conclusion
</h2>

<p>In this article, we explored how to build a real-time chat application using WebSockets and Reactive Streams with Java and Spring Boot. We covered the essential concepts of WebSockets and Reactive Streams, set up a Spring Boot project, and implemented the chat application step-by-step. By combining these technologies, you can build responsive and scalable real-time applications. We encourage you to experiment with these technologies and expand the chat application with additional features like user authentication, private messaging, and message persistence.</p>

DEV Community