Real-Time WebSockets & SSE

# Python — WebSocket hub with heartbeat + Redis backplane (asyncio)
import asyncio, json, contextlib
import websockets
import redis.asyncio as aioredis

HEARTBEAT_S = 25    # MUST be below the LB idle timeout (ALB default: 60s)
PONG_TIMEOUT_S = 10

class Hub:
    def __init__(self):
        self.rooms: dict[str, set] = {}   # room -> local sockets only

    def join(self, room, ws):  self.rooms.setdefault(room, set()).add(ws)
    def leave(self, room, ws): self.rooms.get(room, set()).discard(ws)

    async def deliver_local(self, room, payload):
        # Snapshot the set: sockets can close while we iterate.
        targets = list(self.rooms.get(room, ()))
        if not targets: return
        await asyncio.gather(
            *(self._safe_send(ws, payload) for ws in targets),
            return_exceptions=True)

    async def _safe_send(self, ws, payload):
        with contextlib.suppress(Exception):
            await ws.send(payload)

hub = Hub()

async def backplane(redis):
    # Every node subscribes. A publish on ANY node reaches ALL nodes.
    pubsub = redis.pubsub()
    await pubsub.psubscribe("room:*")
    async for msg in pubsub.listen():
        if msg["type"] != "pmessage": continue
        room = msg["channel"].decode().split(":", 1)[1]
        await hub.deliver_local(room, msg["data"].decode())

async def heartbeat(ws):
    # TCP will NOT tell us the peer vanished. We must probe.
    while True:
        await asyncio.sleep(HEARTBEAT_S)
        try:
            await asyncio.wait_for(await ws.ping(), PONG_TIMEOUT_S)
        except (asyncio.TimeoutError, Exception):
            await ws.close(code=1011, reason="heartbeat timeout")
            return

async def handler(ws):
    redis = aioredis.from_url("redis://localhost")
    room = "lobby"
    hub.join(room, ws)
    hb = asyncio.create_task(heartbeat(ws))
    try:
        async for raw in ws:
            # Publish to the bus, NOT to local sockets — other nodes need it too.
            await redis.publish(f"room:{room}", raw)
    except websockets.ConnectionClosed:
        pass
    finally:
        hb.cancel()
        hub.leave(room, ws)

async def main():
    redis = aioredis.from_url("redis://localhost")
    asyncio.create_task(backplane(redis))
    async with websockets.serve(handler, "0.0.0.0", 8080,
                                  ping_interval=None):  # we ping manually
        await asyncio.Future()

asyncio.run(main())

// C++ — uWebSockets: an epoll event loop, ~1 thread per CPU core.
// This is the architecture that actually reaches C1M on one box.
#include <App.h>
#include <string_view>
#include <chrono>

struct PerSocket {
    std::string room;
    std::chrono::steady_clock::time_point lastPong;
};

int main() {
    uWS::App()
      .ws<PerSocket>("/ws", {
        // Backpressure limit: if a slow client cannot drain this fast enough,
        // the library closes it instead of letting our RAM grow without bound.
        .maxBackpressure  = 1 * 1024 * 1024,
        .maxPayloadLength = 16 * 1024,

        // The library sends PING frames for us and closes on missing PONG.
        .idleTimeout      = 32,   // seconds — keep under the LB's 60s
        .sendPingsAutomatically = true,

        .open = [](auto *ws) {
            ws->getUserData()->room = "lobby";
            // Native pub/sub — O(1) topic fan-out inside this loop's sockets.
            ws->subscribe("room/lobby");
        },

        .message = [](auto *ws, std::string_view msg, uWS::OpCode op) {
            // publish() reaches every socket on THIS loop subscribed to the topic.
            // For multi-NODE fan-out you still forward to Redis/NATS here.
            ws->publish("room/lobby", msg, op);
        },

        .drain = [](auto *ws) {
            // Socket buffer freed up. Resume any paused producer here.
        },

        .close = [](auto *ws, int code, std::string_view) {
            // Unsubscribe is automatic. Release app resources only.
        }
      })
      .listen(8080, [](auto *token) {
          if (!token) throw std::runtime_error("listen failed");
      })
      .run();
}

// Before deploying, raise the OS ceilings or none of this matters:
//   ulimit -n 1048576                       # file descriptors = sockets
//   sysctl -w fs.file-max=2097152
//   sysctl -w net.ipv4.tcp_rmem="4096 16384 262144"   # shrink per-socket RAM

// Java — Spring WebSocket handler with heartbeat + Redis backplane
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import org.springframework.data.redis.core.StringRedisTemplate;

public class ChatHandler extends TextWebSocketHandler {

    // Local sockets ONLY. Cross-node delivery goes through Redis.
    private final Map<String, Set<WebSocketSession>> rooms = new ConcurrentHashMap<>();
    private final StringRedisTemplate redis;
    private final ScheduledExecutorService beat =
        Executors.newSingleThreadScheduledExecutor();

    public ChatHandler(StringRedisTemplate redis) {
        this.redis = redis;
        // One shared timer beats every socket — never one thread per connection.
        beat.scheduleAtFixedRate(this::pingAll, 25, 25, TimeUnit.SECONDS);
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession s) {
        s.setTextMessageSizeLimit(64 * 1024);
        rooms.computeIfAbsent("lobby", k -> ConcurrentHashMap.newKeySet()).add(s);
    }

    @Override
    protected void handleTextMessage(WebSocketSession s, TextMessage m) {
        // Publish to the bus so OTHER nodes' clients also receive it.
        redis.convertAndSend("room:lobby", m.getPayload());
    }

    // Called by the Redis listener on EVERY node, including the publisher.
    public void deliverLocal(String room, String payload) {
        for (WebSocketSession s : rooms.getOrDefault(room, Set.of())) {
            try {
                // A WebSocketSession is NOT thread-safe: serialise writes per socket.
                synchronized (s) {
                    if (s.isOpen()) s.sendMessage(new TextMessage(payload));
                }
            } catch (Exception e) { closeQuietly(s, room); }
        }
    }

    private void pingAll() {
        var probe = new PingMessage();
        rooms.forEach((room, set) -> set.forEach(s -> {
            try { synchronized (s) { if (s.isOpen()) s.sendMessage(probe); } }
            catch (Exception e) { closeQuietly(s, room); }
        }));
    }

    @Override
    public void afterConnectionClosed(WebSocketSession s, CloseStatus st) {
        rooms.values().forEach(set -> set.remove(s));
    }

    private void closeQuietly(WebSocketSession s, String room) {
        rooms.getOrDefault(room, Set.of()).remove(s);
        try { s.close(CloseStatus.SERVER_ERROR); } catch (Exception ignored) {}
    }
}

// Go — the canonical hub pattern. One writer goroutine per socket is
// MANDATORY: concurrent WriteMessage calls on one conn corrupt the stream.
package main

import (
    "net/http"
    "time"

    "github.com/gorilla/websocket"
)

const (
    pongWait   = 60 * time.Second
    pingPeriod = (pongWait * 9) / 10 // 54s — always ping BEFORE the deadline
    writeWait  = 10 * time.Second
)

type Client struct {
    conn *websocket.Conn
    send chan []byte // buffered: absorbs bursts, and its depth IS your backpressure limit
    hub  *Hub
}

type Hub struct {
    clients    map[*Client]bool
    broadcast  chan []byte
    register   chan *Client
    unregister chan *Client
}

// One goroutine owns the map, so no mutex is needed anywhere.
func (h *Hub) Run() {
    for {
        select {
        case c := <-h.register:
            h.clients[c] = true
        case c := <-h.unregister:
            if _, ok := h.clients[c]; ok {
                delete(h.clients, c)
                close(c.send)
            }
        case msg := <-h.broadcast:
            for c := range h.clients {
                select {
                case c.send <- msg:
                default:
                    // Buffer full = client too slow. Drop it rather than block
                    // the hub and stall EVERY other client. Critical detail.
                    delete(h.clients, c)
                    close(c.send)
                }
            }
        }
    }
}

func (c *Client) readPump() {
    defer func() { c.hub.unregister <- c; c.conn.Close() }()

    c.conn.SetReadLimit(64 << 10)
    c.conn.SetReadDeadline(time.Now().Add(pongWait))
    // Each PONG pushes the deadline out. No pong -> the read fails -> cleanup.
    c.conn.SetPongHandler(func(string) error {
        return c.conn.SetReadDeadline(time.Now().Add(pongWait))
    })

    for {
        _, msg, err := c.conn.ReadMessage()
        if err != nil {
            return
        }
        c.hub.broadcast <- msg
    }
}

func (c *Client) writePump() {
    ticker := time.NewTicker(pingPeriod)
    defer func() { ticker.Stop(); c.conn.Close() }()

    for {
        select {
        case msg, ok := <-c.send:
            c.conn.SetWriteDeadline(time.Now().Add(writeWait))
            if !ok { // hub closed the channel
                c.conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }
            if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
                return
            }
        case <-ticker.C:
            c.conn.SetWriteDeadline(time.Now().Add(writeWait))
            if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return // peer is gone; reclaim this connection
            }
        }
    }
}

// ── SERVER (Node + ws): heartbeat that reaps half-open sockets ──
import { WebSocketServer, WebSocket } from 'ws';

const wss = new WebSocketServer({ port: 8080, maxPayload: 64 * 1024 });

interface Live extends WebSocket { isAlive?: boolean; }

wss.on('connection', (ws: Live) => {
  ws.isAlive = true;
  ws.on('pong', () => { ws.isAlive = true; });   // the liveness proof

  ws.on('message', (raw) => {
    // Guard against slow consumers: bufferedAmount is unsent bytes.
    for (const peer of wss.clients) {
      if (peer.readyState === WebSocket.OPEN && peer.bufferedAmount < 1_000_000) {
        peer.send(raw, { binary: false });
      }
    }
  });
});

// ONE interval sweeps every socket. 25s stays under the ALB's 60s idle timeout.
const sweep = setInterval(() => {
  for (const ws of wss.clients as Set<Live>) {
    if (ws.isAlive === false) { ws.terminate(); continue; } // missed last pong
    ws.isAlive = false;
    ws.ping();
  }
}, 25_000);
wss.on('close', () => clearInterval(sweep));

// ── CLIENT: reconnect with EXPONENTIAL BACKOFF + JITTER ──
// Without the jitter, 50k clients reconnect in lockstep and DDoS you.
class ReconnectingSocket {
  private attempt = 0;
  private ws?: WebSocket;

  constructor(private url: string, private onMsg: (d: string) => void) {
    this.connect();
  }

  private connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => { this.attempt = 0; };      // reset the ladder
    this.ws.onmessage = (e) => this.onMsg(e.data);
    this.ws.onclose = (e) => {
      if (e.code === 1000) return;                  // clean, intentional close
      const base = 500, cap = 30_000;
      const backoff = Math.min(cap, base * 2 ** this.attempt++);
      const jittered = backoff * (0.5 + Math.random());  // ← the vital part
      setTimeout(() => this.connect(), jittered);
    };
  }
}

// ── Compare: SSE needs NONE of that reconnect code ──
const es = new EventSource('/stream');
es.addEventListener('price', (e) => render(JSON.parse(e.data)));
// The browser reconnects automatically AND replays via Last-Event-ID.