Connection & Registry

What the interviewer just asked

“Show me how a user connects and how you track that connection.”

This is where Spring’s SseEmitter meets your three-map registry. Implement the controller shell, the service fields, connect(), and disconnect() before touching watch or publish.

One SSE connection per user

Each user establishes exactly one SSE connection when they open the page. All stock price updates for all their watched symbols flow through this single connection. The SSE event’s name field carries the symbol so the client knows which price belongs to which stock.

@RestController
public class StockController {

    private final StockService stockService;

    @GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter connect(@RequestParam String userId) {
        // SseEmitter is a write-handle to this user's open HTTP connection.
        // Returning it tells Spring: keep this response open and write to it
        // whenever emitter.send() is called. If we don't return it, Spring
        // closes the response immediately and emitter.send() has nowhere to write.
        return stockService.connect(userId);
    }

    @PostMapping("/watch")
    public void addSymbol(@RequestParam String userId, @RequestParam String symbol) {
        stockService.addSymbol(userId, symbol);
    }

    @DeleteMapping("/watch")
    public void removeSymbol(@RequestParam String userId, @RequestParam String symbol) {
        stockService.removeSymbol(userId, symbol);
    }

    @PutMapping("/watch")
    public void replaceSymbols(@RequestParam String userId, @RequestBody List<String> symbols) {
        stockService.replaceSymbols(userId, symbols);
    }
}

The controller is thin — all registry logic lives in StockService. That is deliberate: the interviewer may ask you to add publish or swap SSE for WebSocket without touching the REST surface.

The three maps as service fields

@Service
public class StockService {

    // one live SSE connection per user
    private final Map<String, SseEmitter> userEmitters = new ConcurrentHashMap<>();

    // which symbols each user is currently watching
    private final Map<String, Set<String>> userSymbols = new ConcurrentHashMap<>();

    // which users are watching each symbol — used during fan-out
    private final Map<String, Set<String>> symbolUsers = new ConcurrentHashMap<>();
}

All three are ConcurrentHashMap because connect, watch, disconnect, and publish can be invoked from different threads concurrently — Spring serves HTTP requests on a thread pool, and the upstream price feed runs on its own thread.

connect() — opening the SSE connection

public SseEmitter connect(String userId) {
    SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);

    // if user reconnects (e.g. page refresh), complete the old emitter first
    SseEmitter existing = userEmitters.put(userId, emitter);
    if (existing != null) existing.complete();

    // initialise symbol set for this user if not already present
    userSymbols.putIfAbsent(userId, ConcurrentHashMap.newKeySet());

    // clean up fully when the connection closes
    emitter.onCompletion(() -> disconnect(userId));
    emitter.onTimeout(()    -> disconnect(userId));
    emitter.onError(e       -> disconnect(userId));

    return emitter;
}

Walk through each line on the board:

new SseEmitter(Long.MAX_VALUE) — creates the write handle with no timeout (or set a reasonable timeout in production). Spring keeps the HTTP response open until complete() is called or the client disconnects.

userEmitters.put(userId, emitter) — registers this connection. If the user refreshes the page, a new connect arrives while the old connection may still be open. Completing the old emitter closes the stale HTTP response and triggers its cleanup callback.

userSymbols.putIfAbsent(userId, ConcurrentHashMap.newKeySet()) — ensures the user has a symbol set ready before any watch call. putIfAbsent is safe under concurrent connect + watch.

Callbacks → disconnect(userId) — whichever way the connection ends (normal close, timeout, error), the same cleanup runs. Without these callbacks, disconnected users linger in symbolUsers and receive fan-out attempts on dead emitters.

ConcurrentHashMap.newKeySet()

You will use this in connect, add, remove, and replace. It deserves a sentence on the board:

ConcurrentHashMap.newKeySet() returns a thread-safe Set backed by a ConcurrentHashMap — the right choice for a set that is read during fan-out and written during add/remove.

Why not Collections.synchronizedSet(new HashSet<>())? That wraps every operation in one lock — fan-out iterating the set blocks add/remove and vice versa. newKeySet() allows concurrent iteration and mutation with finer-grained locking, which matters when publish walks symbolUsers["AAPL"] while another thread calls addSymbol for AAPL.

Why not a plain HashSet inside a ConcurrentHashMap? The outer map is thread-safe, but the inner set is not. Two threads calling symbols.add("AAPL") on a plain HashSet can corrupt internal state.

disconnect() — full cleanup on connection close

When a user’s connection closes (browser tab closed, network drop, page navigation), everything associated with that user must be cleaned up:

private void disconnect(String userId) {
    // remove the emitter
    userEmitters.remove(userId);

    // remove this user from every symbol they were watching
    Set<String> symbols = userSymbols.remove(userId);
    if (symbols != null) {
        for (String symbol : symbols) {
            Set<String> users = symbolUsers.get(symbol);
            if (users != null) users.remove(userId);
        }
    }
}

This is the disconnect traversal path from the previous chapter:

disconnect("U1")
  │
  ├─ userEmitters.remove("U1")           emitter gone
  │
  └─ userSymbols.remove("U1") → {AAPL, TSLA}
       │
       ├─ symbolUsers["AAPL"].remove("U1")
       └─ symbolUsers["TSLA"].remove("U1")

After disconnect(), no state for this user remains in any of the three maps. The next publish() for any of their former symbols will not find them.

Do not skip the symbolUsers cleanup. If you only remove the emitter, symbolUsers["AAPL"] still contains U1. The next publish will try emitter.send() on a removed emitter — exceptions, log spam, or worse, a memory leak of dead user IDs in every symbol set.

Reconnect handling

Page refresh produces this sequence:

1. Browser closes old SSE connection
2. onCompletion → disconnect(U1)        old state cleaned up
3. Browser opens GET /connect?userId=U1
4. connect(U1)                          fresh emitter, empty symbol set
5. POST /watch calls                    re-register symbols

If step 2 is slow and step 3 arrives first, connect completes the old emitter via existing.complete(), which triggers step 2’s callback. Either way, you end with one live emitter and consistent maps.

What the user sees after connecting

The connection is open but no prices arrive yet — the user has not watched any symbols. The client must call POST /watch to start receiving updates. Typically the frontend sends the initial watchlist immediately after establishing the SSE connection:

1. GET /connect?userId=U1          → SSE connection open
2. POST /watch?userId=U1&symbol=AAPL
3. POST /watch?userId=U1&symbol=TSLA
   ... (10 more)
4. data: {"symbol":"AAPL","price":185.42}   ← prices start flowing

State after step 1:

userEmitters:  { U1 → emitter1 }
userSymbols:   { U1 → {} }          empty set, ready for watch calls
symbolUsers:   { }                  nothing registered yet

State after steps 2–3 (AAPL and TSLA):

userEmitters:  { U1 → emitter1 }
userSymbols:   { U1 → {AAPL, TSLA} }
symbolUsers:   { AAPL → {U1}, TSLA → {U1} }

Now publish("AAPL", price) will find U1 and push to emitter1.

Common interviewer follow-ups

“What if connect is called twice without disconnect?” The second call replaces the emitter in userEmitters and completes the first. The first emitter’s callback also fires disconnect — idempotent cleanup on an already-replaced user is safe because remove on a missing key is a no-op.

“Why not store symbols inside the emitter somehow?” SseEmitter is Spring’s transport primitive — it has no subscription API. Keeping subscription state in your maps separates concerns and makes fan-out testable without a live HTTP connection.

“What timeout should you use?” Long.MAX_VALUE is fine for the interview. In production, set a timeout aligned with your load balancer idle timeout and rely on client reconnect with exponential backoff.