package com.example.notifications;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
@Component
public class SseBroadcaster {
private static final long STREAM_TIMEOUT_MS = 30 * 60 * 1000L;
private final List<SseEmitter> emitters = new CopyOnWriteArrayList<>();
public SseEmitter subscribe() {
SseEmitter emitter = new SseEmitter(STREAM_TIMEOUT_MS);
emitters.add(emitter);
emitter.onCompletion(() -> emitters.remove(emitter));
emitter.onTimeout(() -> emitters.remove(emitter));
emitter.onError(e -> emitters.remove(emitter));
return emitter;
}
public void broadcast(String eventName, String eventId, Object payload) {
for (SseEmitter emitter : emitters) {
try {
emitter.send(SseEmitter.event()
.id(eventId)
.name(eventName)
.data(payload));
} catch (IOException ex) {
emitter.completeWithError(ex);
}
}
}
public int activeConnections() {
return emitters.size();
}
}
package com.example.notifications;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
@RestController
@RequestMapping("/notifications")
public class NotificationController {
private final SseBroadcaster broadcaster;
public NotificationController(SseBroadcaster broadcaster) {
this.broadcaster = broadcaster;
}
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter stream() {
SseEmitter emitter = broadcaster.subscribe();
try {
emitter.send(SseEmitter.event()
.name("connected")
.data("stream opened"));
} catch (IOException ex) {
emitter.completeWithError(ex);
}
return emitter;
}
}
package com.example.notifications;
import org.springframework.stereotype.Service;
import java.time.Instant;
import java.util.UUID;
@Service
public class NotificationService {
private final SseBroadcaster broadcaster;
public NotificationService(SseBroadcaster broadcaster) {
this.broadcaster = broadcaster;
}
public void notifyUser(long userId, String title, String message) {
String id = UUID.randomUUID().toString();
Notification payload = new Notification(id, userId, title, message, Instant.now());
broadcaster.broadcast("notification", id, payload);
}
public record Notification(String id,
long userId,
String title,
String message,
Instant createdAt) {
}
}
Server-Sent Events (SSE) provide a lightweight, one-directional streaming channel from server to browser over a single long-lived HTTP connection. Unlike WebSockets, SSE rides on plain HTTP, auto-reconnects natively via the browser's EventSource, and needs no extra protocol handshake, which makes it ideal for pushing notifications, progress updates, or activity feeds. This snippet shows the full server side: a broadcaster that owns the live connections, a controller that hands the browser a stream, and a service that publishes domain events onto it.
In SseBroadcaster, each subscriber is represented by a Spring SseEmitter stored in a CopyOnWriteArrayList. That collection is chosen deliberately: reads (iterating during a broadcast) vastly outnumber writes (subscribe/unsubscribe), and it avoids ConcurrentModificationException when an emitter is removed mid-iteration. The subscribe method registers cleanup callbacks — onCompletion, onTimeout, and onError all deregister the emitter so dead connections don't leak. A generous timeout is passed in so the browser holds the stream open rather than reconnecting every 30 seconds. Sending is wrapped in a try/catch: if a client has gone away, emitter.send throws IOException, and the code calls completeWithError to trigger the same cleanup path.
The NotificationController exposes GET /notifications/stream, which returns the raw SseEmitter produced by the broadcaster. Spring recognizes the return type and keeps the request thread's response open for streaming. Crucially it sends an initial event("connected") immediately so proxies and the browser confirm the stream is live before any real data arrives. The produces = TEXT_EVENT_STREAM_VALUE media type is what tells clients and intermediaries this is an SSE response.
In NotificationService, business logic like notifyUser builds a Notification payload and delegates to the broadcaster. Named events (.name("notification")) let the front-end register targeted addEventListener handlers instead of a generic catch-all. Providing a stable .id(...) lets the browser send Last-Event-ID on reconnect so a more advanced implementation could replay missed messages.
The main pitfalls to remember: SSE is unidirectional, most browsers cap concurrent connections per domain, and a load-balanced deployment needs sticky sessions or an external fan-out (Redis pub/sub) because emitters live in one JVM's memory. For a single service pushing to modest numbers of clients, this pattern is simple and robust.
Related snips
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
type User {
id: ID!
name: String!
email: String!
posts: [Post!]!
createdAt: String!GraphQL API with Spring Boot
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
