Implementing Real-Time Data Streaming with Spring's SseEmitter

SseEmitter is a utility class provided by the Spring Framework that enables the implementation of Server-Sent Events (SSE) in Spring MVC or Spring Boot applications. SSE allows servers to push real-time data to clients (typically browsers) through unidirectional, persistent connections, eliminating the need for clients to repeatedly make requests.

Server-Side Implementation

1. Creating an SseEmitter Instance

In your HTTP request handler method, create an SseEmitter instance. Typically, you'll associate this instance with a client identifier (such as user ID or specific resource ID) to target specific clients when pushing messages.

@GetMapping("/realtime-data")
public SseEmitter establishDataStream() {
    SseEmitter emitter = new SseEmitter(); // Optional: set timeout, e.g., new SseEmitter(30000L);
    return emitter;
}

2. Configuring SseEmitter

You can configure properties such as timeout duration and heartbeat intervals to maintain an active connection or close it when necessary.

3. Handling Data Push

When new data needs to be sent to clients, call the SseEmitter's send() method. You can send plain text messages, data objects (serialized to JSON or other formats), and even custom event types.

public void broadcastUpdate(String content) {
    try {
        emitter.send(SseEmitter.event().data(content));
    } catch (IOException e) {
        // Handle exception, clean up resources, and notify other systems about the disconnection
        emitter.completeWithError(e);
    }
}

4. Managing Connection Lifecycle

Ensure proper resource cleanup and client notification when connections are disconnected, exceptions occur, or application logic requires closing connections. This typically involves registering listeners for SseEmitter's onError() and onCompletion() events.

emitter.onCompletion(() -> {
    // Clean up resources associated with this client
    ACTIVE_CONNECTIONS.remove(clientId);
});

emitter.onError((ex) -> {
    // Handle error, such as retrying or logging
    emitter.completeWithError(ex);
    ACTIVE_CONNECTIONS.remove(clientId);
});

5. Storing and Managing SseEmitter Instances

If you need to push data to multiple clients, you may need to store SseEmitter instances in a collection (like a Map) for later retrieval to send messages to the appropriate client. This is usually associated with a unique client identifier.

private static final Map<String, SseEmitter> ACTIVE_CONNECTIONS = new ConcurrentHashMap<>();

@GetMapping("/subscribe/{clientId}")
public SseEmitter createSubscription(@PathVariable String clientId) {
    SseEmitter emitter = new SseEmitter();
    ACTIVE_CONNECTIONS.put(clientId, emitter);
    // Add lifecycle listeners and data push logic
    return emitter;
}

Client-Side Implementation

1. Creating an EventSource Object

Use JavaScript in the browser to create an EventSource object, specifying the SSE resource URL provided by the server.

const eventSource = new EventSource('/realtime-data');

2. Handling Received Events

Register event handlers to process different types of events pushed from the server. SSE typically uses the 'message' event to transmit data, but custom event types can also be implemented.

eventSource.addEventListener('message', function(event) {
    const payload = event.data; // Received data
    console.log('Data received:', payload);
    // Update UI or perform other business logic
});

// Optional: Handle custom events
eventSource.addEventListener('notification', function(event) {
    const data = event.data;
    // ...
});

3. Handling Connection State Changes

SSE automatically handles reconnection on disconnection, but you can also listen to 'open' and 'error' events to respond to connection state changes.

eventSource.addEventListener('open', function(event) {
    console.log('Connection established');
});

eventSource.addEventListener('error', function(event) {
    if (event.readyState === EventSource.CLOSED) {
        console.error('Connection closed');
        // May need manual reconnection or other recovery strategies
    }
});