@RestController
@RequestMapping("/api/notifications")
public class NotificationController {
private final SseEmitterRegistry registry;
private final NotificationService service;
public NotificationController(SseEmitterRegistry registry, NotificationService service) {
this.registry = registry;
this.service = service;
}
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter stream(@AuthenticationPrincipal CurrentUser user) {
SseEmitter emitter = registry.subscribe(user.getId());
try {
emitter.send(SseEmitter.event().name("connected").data(Map.of("userId", user.getId())));
} catch (IOException ex) {
emitter.completeWithError(ex);
}
return emitter;
}
@PostMapping
@ResponseStatus(HttpStatus.ACCEPTED)
public void publish(@AuthenticationPrincipal CurrentUser user,
@Valid @RequestBody NotificationRequest request) {
service.deliver(user.getId(), request.getTargetUserId(), request.getMessage());
}
}@Service
public class NotificationService {
private final NotificationRepository repository;
private final SseEmitterRegistry registry;
public NotificationService(NotificationRepository repository, SseEmitterRegistry registry) {
this.repository = repository;
this.registry = registry;
}
@Transactional
public Notification deliver(Long senderId, Long targetUserId, String message) {
Notification notification = new Notification();
notification.setSenderId(senderId);
notification.setUserId(targetUserId);
notification.setMessage(message);
notification.setCreatedAt(Instant.now());
Notification saved = repository.save(notification);
registry.send(targetUserId, "notification", NotificationView.from(saved));
return saved;
}
}@Component
public class SseEmitterRegistry {
private static final long TIMEOUT = 30 * 60 * 1000L;
private final Map<Long, List<SseEmitter>> emitters = new ConcurrentHashMap<>();
public SseEmitter subscribe(Long userId) {
SseEmitter emitter = new SseEmitter(TIMEOUT);
emitters.computeIfAbsent(userId, id -> new CopyOnWriteArrayList<>()).add(emitter);
emitter.onCompletion(() -> remove(userId, emitter));
emitter.onTimeout(() -> remove(userId, emitter));
emitter.onError(e -> remove(userId, emitter));
return emitter;
}
public void send(Long userId, String eventName, Object payload) {
List<SseEmitter> userEmitters = emitters.get(userId);
if (userEmitters == null) {
return;
}
for (SseEmitter emitter : userEmitters) {
try {
emitter.send(SseEmitter.event().name(eventName).data(payload));
} catch (IOException | IllegalStateException ex) {
remove(userId, emitter);
}
}
}
private void remove(Long userId, SseEmitter emitter) {
List<SseEmitter> userEmitters = emitters.get(userId);
if (userEmitters != null) {
userEmitters.remove(emitter);
if (userEmitters.isEmpty()) {
emitters.remove(userId);
}
}
}
}This snippet shows how live user notifications are pushed to browsers using Server-Sent Events (SSE) in a plain Spring MVC application, without a full WebSocket stack. SSE is a good fit here because notifications flow one way (server to client), rely on a long-lived HTTP response, and get automatic reconnection from the browser's EventSource for free.
The SseEmitterRegistry tab is the heart of the design. It holds a ConcurrentHashMap keyed by user id, where each value is a CopyOnWriteArrayList<SseEmitter> so one user can have several open tabs or devices. On subscribe, it constructs an SseEmitter with a long timeout and wires onCompletion, onTimeout, and onError callbacks so the emitter removes itself from the registry when the connection drops. This self-cleanup is the critical part: without it, dead emitters accumulate and every future send throws. The send method iterates a user's emitters, calls emitter.send(...) with a named event and payload, and on IOException marks the emitter stale so it is pruned. CopyOnWriteArrayList makes concurrent iteration during a broadcast safe against subscribers connecting or disconnecting mid-loop.
The NotificationController tab exposes two endpoints. The stream handler returns an SseEmitter with produces = text/event-stream; Spring keeps the request thread's response open and streams asynchronously. It immediately sends a small connected event so clients and proxies confirm the pipe is live and the browser resets its retry timer. The publish endpoint accepts a notification body and fans it out through the registry to the target user.
The NotificationService tab sits between domain logic and transport: it persists the notification, then calls registry.send so delivery is decoupled from any single request. Because SSE only reaches currently-connected clients, persistence ensures a user who was offline still sees the notification on next load.
A few trade-offs matter. SSE runs over one HTTP connection per client, so heavy fan-out benefits from an async request timeout and enough server threads or a reactive stack. Proxies may buffer text/event-stream, so disabling response buffering is often required. For multi-instance deployments this in-memory registry must be backed by a shared bus like Redis pub/sub, since an emitter only exists on the node that accepted the connection. This pattern is ideal for dashboards, toasts, and progress feeds where WebSockets would be overkill.
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.
