java 60 lines · 8 steps

Server-Sent Events per user in Spring

A Spring controller that streams live notifications to each user over SSE, tracking open connections in a thread-safe map.

Explained by highlit

Spring Advanced server-sent-events concurrency event-driven streaming resource-cleanup

1@RestController
2@RequestMapping("/api/notifications")
3public class NotificationStreamController {
4 
  private final Map<String, CopyOnWriteArrayList<SseEmitter>> subscribers = new ConcurrentHashMap<>();
6 
  @GetMapping(value = "/stream/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
  public SseEmitter subscribe(@PathVariable String userId) {
      SseEmitter emitter = new SseEmitter(Duration.ofMinutes(30).toMillis());
10 
      subscribers.computeIfAbsent(userId, key -> new CopyOnWriteArrayList<>()).add(emitter);
12 
      emitter.onCompletion(() -> removeEmitter(userId, emitter));
      emitter.onTimeout(() -> {
          emitter.complete();
          removeEmitter(userId, emitter);
      });
      emitter.onError(ex -> removeEmitter(userId, emitter));
19 
      try {
          emitter.send(SseEmitter.event()
                  .name("connected")
                  .data(Map.of("userId", userId, "since", Instant.now().toString())));
      } catch (IOException ex) {
          emitter.completeWithError(ex);
      }
27 
      return emitter;
  }
30 
  @EventListener
  public void onNotification(NotificationCreatedEvent event) {
      CopyOnWriteArrayList<SseEmitter> emitters = subscribers.get(event.userId());
      if (emitters == null) {
          return;
      }
      for (SseEmitter emitter : emitters) {
          try {
              emitter.send(SseEmitter.event()
                      .id(event.id())
                      .name("notification")
                      .reconnectTime(3000)
                      .data(event.payload(), MediaType.APPLICATION_JSON));
          } catch (IOException ex) {
              emitter.completeWithError(ex);
              removeEmitter(event.userId(), emitter);
          }
      }
  }
50 
  private void removeEmitter(String userId, SseEmitter emitter) {
      CopyOnWriteArrayList<SseEmitter> emitters = subscribers.get(userId);
      if (emitters != null) {
          emitters.remove(emitter);
          if (emitters.isEmpty()) {
              subscribers.remove(userId);
          }
      }
  }
60}

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1SseEmitter keeps a one-way HTTP connection open so the server can push events until it completes or times out.
2Registering completion, timeout, and error callbacks is essential to prevent leaking dead connections from your registry.
3Decoupling delivery from creation via an application event lets any part of the system fan out messages to connected clients.

Related explainers

java

public class SlidingLogRateLimiter {
 
    private final int maxRequests;
    private final long windowMillis;

How a sliding-log rate limiter works

rate-limiting concurrency sliding-window

Advanced 8 steps

package theme
 
import (
	"fmt"

Per-tenant HTML templates with Gin's renderer

multi-tenancy concurrency html-templates

Advanced 8 steps

php

<?php
 
namespace App\Http\Controllers;

Streaming a filtered CSV export in Laravel

streaming csv-export query-builder

Intermediate 9 steps

python

import time
import threading
from enum import Enum
from functools import wraps

Building a circuit breaker in Python

circuit-breaker resilience decorators

Advanced 7 steps

java

public final class Slugifier {
 
    private static final Pattern NON_LATIN = Pattern.compile("[^\\w-]");
    private static final Pattern WHITESPACE = Pattern.compile("[\\s]+");

Building a URL slugifier in Java

regex unicode-normalization string-processing

Intermediate 8 steps

typescript

import { Injectable } from '@angular/core';
import { HttpInterceptor, HttpRequest, HttpHandler, HttpEvent, HttpErrorResponse } from '@angular/common/http';
import { BehaviorSubject, Observable, throwError } from 'rxjs';
import { catchError, filter, take, switchMap } from 'rxjs/operators';