Real Time Communication Patterns

Real-Time Communication Patterns

Overview

Real-time communication introduces connection lifecycle complexity, horizontal scaling challenges, and security risks that request/response APIs avoid entirely. A single unbounded connection pool, missing heartbeat, or JWT exposed in a query string can bring down production systems.

When to use: Designing live dashboards, chat, collaborative editing, notifications, streaming APIs, IoT device feeds, or any feature where the server must push data to clients unprompted.

Quick Reference

| Protocol | Direction | Reconnect | Scaling | Best For | |----------|-----------|-----------|---------|----------| | WebSocket | Full-duplex | Manual | Redis Pub/Sub or sticky sessions | Chat, games, collaborative editing | | Server-Sent Events (SSE) | Server → client | Auto (EventSource) | Standard HTTP load balancer | Notifications, live feeds, dashboards | | Long Polling | Server → client | Per-request | Standard HTTP load balancer | Legacy clients, firewall-constrained envs | | gRPC Streaming | Full-duplex | Manual | L7 load balancer (Envoy) | Microservice-to-microservice real-time |

---

Patterns in Detail

1. WebSocket Lifecycle

A WebSocket connection passes through distinct phases: HTTP upgrade handshake, open/connected, heartbeat keep-alive, graceful close, and reconnection with backoff.

Red Flags:

• No heartbeat — idle connections silently dropped by load balancers or NAT devices
• No reconnection logic — network blips kill the session permanently
• Sending frames after close — race condition on the write path
• Missing onclose / onerror handlers — uncaught errors crash the page or goroutine

TypeScript — client lifecycle with reconnect:

interface WsOptions {
  url: string;
  heartbeatIntervalMs?: number;
  maxReconnectDelayMs?: number;
  onMessage: (data: unknown) => void;
}

class ManagedWebSocket {
  private ws: WebSocket | null = null;
  private heartbeatTimer: ReturnType | null = null;
  private reconnectDelay = 1_000;

  constructor(private readonly opts: WsOptions) {
    this.connect();
  }

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

    this.ws.onopen = () => {
      this.reconnectDelay = 1_000; // reset on successful connect
      this.startHeartbeat();
    };

    this.ws.onmessage = (event) => {
      const msg = JSON.parse(event.data as string) as { type: string; payload: unknown };
      if (msg.type === 'pong') return; // heartbeat response — ignore
      this.opts.onMessage(msg.payload);
    };

    this.ws.onclose = (event) => {
      this.stopHeartbeat();
      if (!event.wasClean) this.scheduleReconnect();
    };

    this.ws.onerror = (err) => {
      console.error('[WS] error', err);
      this.ws?.close();
    };
  }

  private startHeartbeat(): void {
    const interval = this.opts.heartbeatIntervalMs ?? 25_000;
    this.heartbeatTimer = setInterval(() => {
      if (this.ws?.readyState === WebSocket.OPEN) {
        this.ws.send(JSON.stringify({ type: 'ping' }));
      }
    }, interval);
  }

  private stopHeartbeat(): void {
    if (this.heartbeatTimer) clearInterval(this.heartbeatTimer);
    this.heartbeatTimer = null;
  }

  private scheduleReconnect(): void {
    const maxDelay = this.opts.maxReconnectDelayMs ?? 30_000;
    const jitter = Math.random() * 1_000;
    const delay = Math.min(this.reconnectDelay + jitter, maxDelay);
    this.reconnectDelay = Math.min(this.reconnectDelay * 2, maxDelay);
    setTimeout(() => this.connect(), delay);
  }

  send(payload: unknown): void {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(payload));
    }
  }

  close(): void {
    this.stopHeartbeat();
    this.ws?.close(1000, 'Client closing gracefully');
  }
}

Go — server-side ping/pong with gorilla/websocket:

import (
    "time"
    "github.com/gorilla/websocket"
)

const (
    writeWait      = 10 * time.Second
    pongWait       = 60 * time.Second
    pingInterval   = (pongWait * 9) / 10
    maxMessageSize = 512 * 1024 // 512 KB
)

func handleConn(conn *websocket.Conn, outbound  {
    res.write(`id: ${event.id}\n`);
    res.write(`event: ${event.type}\n`);
    res.write(`data: ${JSON.stringify(event.data)}\n\n`);
  };

  bus.on('event', listener);
  req.on('close', () => bus.off('event', listener));
}

function replayMissedEvents(afterId: string, res: Response): void {
  // fetch from durable store and stream missed events before live ones
}

Go — net/http SSE endpoint:

func sseHandler(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "SSE not supported", http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("X-Accel-Buffering", "no")

    lastID := r.Header.Get("Last-Event-ID")
    _ = lastID // replay logic goes here

    fmt.Fprintf(w, "retry: 5000\n\n")
    flusher.Flush()

    for {
        select {
        case  {
  while (true) {
    try {
      const res = await fetch(`${endpoint}?after=${lastSeq}`, {
        signal: AbortSignal.timeout(35_000), // server timeout is 30 s
      });
      if (res.status === 204) {
        // timeout with no event — reconnect immediately
        continue;
      }
      const events = await res.json() as Array;
      for (const ev of events) {
        processEvent(ev.payload);
        lastSeq = ev.seq;
      }
    } catch (err) {
      const jitter = Math.random() * 2_000;
      await new Promise(r => setTimeout(r, 2_000 + jitter));
    }
  }
}

Go — server hold-until-event with context cancellation:

func longPollHandler(w http.ResponseWriter, r *http.Request) {
    afterSeq, _ := strconv.ParseInt(r.URL.Query().Get("after"), 10, 64)

    ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
    defer cancel()

    events, err := store.WaitForEvents(ctx, afterSeq)
    if err != nil || len(events) == 0 {
        w.WriteHeader(http.StatusNoContent)
        return
    }

    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(events)
}

Thundering herd mitigation — randomized reconnect delay:

// After server restart or broadcast, add per-client jitter
const baseDelay = 500;
const jitter = Math.floor(Math.random() * 4_500); // 0–4.5 s spread
await new Promise(r => setTimeout(r, baseDelay + jitter));

---

4. Protocol Selection Decision Matrix

| Criterion | WebSocket | SSE | Long Polling | gRPC Streaming | |-----------|-----------|-----|--------------|----------------| | Client sends frequently | Yes | No | No | Yes | | Standard HTTP load balancer | No (requires sticky or L7) | Yes | Yes | No (HTTP/2 required) | | Browser native support | Yes | Yes | Yes | No (grpc-web) | | Auto-reconnect built in | No | Yes | Partial | No | | Binary frames | Yes | No (base64) | No | Yes (protobuf) | | Firewall/proxy friendly | Sometimes | Yes | Yes | Sometimes | | Server pushes rarely | Overkill | Good fit | Good fit | Overkill | | Bidirectional required | Yes | No | No | Yes |

Decision rules:

1. If browser client + server-push only → prefer SSE (simpler, auto-reconnect, standard LB)
2. If bidirectional or high-frequency client messages → use WebSocket
3. If legacy environment, mobile proxy restrictions, or simple notification → use long polling
4. If service-to-service streaming with binary payloads → use gRPC bidirectional streaming

---

5. Connection Manager Pattern

A connection manager maintains a per-user (or per-room/channel) registry of active connections, handles fan-out delivery, and enforces connection limits.

Red Flags:

• Global map without mutex — concurrent map writes panic in Go
• No per-user connection limit — one user opens thousands of tabs and exhausts file descriptors
• Fan-out blocks on slow clients — one unresponsive connection delays all others
• No cleanup on disconnect — dead connections accumulate in the registry

TypeScript — room-based connection manager:

type ConnectionId = string;

interface Connection {
  id: ConnectionId;
  userId: string;
  send: (data: string) => void;
}

class ConnectionManager {
  // room → set of connection IDs
  private rooms = new Map>();
  // connection ID → connection
  private connections = new Map();
  private readonly maxPerUser: number;

  constructor(maxPerUser = 5) {
    this.maxPerUser = maxPerUser;
  }

  register(conn: Connection): void {
    const userConnCount = [...this.connections.values()]
      .filter(c => c.userId === conn.userId).length;
    if (userConnCount >= this.maxPerUser) {
      throw new Error(`Connection limit reached for user ${conn.userId}`);
    }
    this.connections.set(conn.id, conn);
  }

  unregister(connId: ConnectionId): void {
    this.connections.delete(connId);
    for (const members of this.rooms.values()) members.delete(connId);
  }

  join(connId: ConnectionId, room: string): void {
    if (!this.rooms.has(room)) this.rooms.set(room, new Set());
    this.rooms.get(room)!.add(connId);
  }

  leave(connId: ConnectionId, room: string): void {
    this.rooms.get(room)?.delete(connId);
  }

  // fan-out: deliver to all room members concurrently, skip slow/dead connections
  fanOut(room: string, data: string): void {
    const members = this.rooms.get(room);
    if (!members) return;
    for (const connId of members) {
      const conn = this.connections.get(connId);
      if (!conn) { members.delete(connId); continue; }
      try { conn.send(data); }
      catch (err) { console.warn('[CM] send failed, removing', connId, err); this.unregister(connId); }
    }
  }
}

Go — concurrent-safe connection hub:

type Hub struct {
    mu      sync.RWMutex
    rooms   map[string]map[string]*Conn // room → connID → Conn
    conns   map[string]*Conn            // connID → Conn
}

func NewHub() *Hub {
    return &Hub{
        rooms: make(map[string]map[string]*Conn),
        conns: make(map[string]*Conn),
    }
}

func (h *Hub) Register(conn *Conn) {
    h.mu.Lock()
    defer h.mu.Unlock()
    h.conns[conn.ID] = conn
}

func (h *Hub) Unregister(connID string) {
    h.mu.Lock()
    defer h.mu.Unlock()
    delete(h.conns, connID)
    for _, members := range h.rooms {
        delete(members, connID)
    }
}

func (h *Hub) FanOut(room string, data []byte) {
    h.mu.RLock()
    members := h.rooms[room]
    h.mu.RUnlock()

    for _, conn := range members {
        select {
        case conn.Send  {
  await subscriber.subscribe(`room:${room}`, (message) => {
    manager.fanOut(room, message);
  });
}

// Any server instance can publish; Redis broadcasts to all subscribers
async function publishToRoom(room: string, data: unknown): Promise {
  await publisher.publish(`room:${room}`, JSON.stringify(data));
}

Go — Redis Pub/Sub bridge:

func startRedisBridge(rdb *redis.Client, hub *Hub, room string) {
    pubsub := rdb.Subscribe(context.Background(), "room:"+room)
    ch := pubsub.Channel()
    go func() {
        for msg := range ch {
            hub.FanOut(room, []byte(msg.Payload))
        }
    }()
}

Sticky sessions — when Redis is overkill:

• Configure the load balancer to hash by session cookie or user ID so one user always lands on the same pod.
• Acceptable for small deployments; increases blast radius on pod restarts.
• Combine with a Redis adapter for graceful draining: before a pod shuts down, migrate connections to peers.

---

7. Auth at Upgrade

WebSocket and SSE connections are established over HTTP, which means standard auth middleware runs exactly once at upgrade/connection time. After that, the long-lived connection is not re-authenticated on every message.

Red Flags:

• JWT in query string (ws://host/chat?token=eyJ...) — tokens appear in server logs, proxy logs, and browser history
• No expiry check at upgrade — a revoked or expired token keeps its connection alive indefinitely
• Cookie-based auth skipped on the upgrade request — missing credentials: 'include' on the client
• No re-auth for long-lived connections — a 24-hour connection carries a 1-hour token with no renewal

Short-lived ticket pattern (preferred over query param JWT):

// Step 1: client calls a REST endpoint while authenticated to get a one-time ticket
const { ticket } = await fetch('/api/ws-ticket', { method: 'POST' }).then(r => r.json());

// Step 2: use the ticket (not the JWT) in the WebSocket URL — it is short-lived and single-use
const ws = new WebSocket(`wss://host/ws?ticket=${ticket}`);

Go — server-side ticket validation at upgrade:

var upgrader = websocket.Upgrader{
    CheckOrigin: func(r *http.Request) bool {
        return isAllowedOrigin(r.Header.Get("Origin"))
    },
}

func wsHandler(w http.ResponseWriter, r *http.Request) {
    ticket := r.URL.Query().Get("ticket")
    userID, err := ticketStore.Consume(ticket) // single-use, TTL 30 s
    if err != nil {
        http.Error(w, "Unauthorized", http.StatusUnauthorized)
        return
    }

    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        return
    }
    defer conn.Close()

    handleConn(conn, userID)
}

Token refresh for long-lived connections:

// Client sends a refresh message before the JWT expires
setInterval(async () => {
  const newToken = await refreshAccessToken();
  ws.send(JSON.stringify({ type: 'auth-refresh', token: newToken }));
}, TOKEN_TTL_MS * 0.8);

---

8. Backpressure and Flow Control

A fast producer with a slow consumer causes unbounded buffer growth, out-of-memory crashes, or stale data delivery. Backpressure strategies prevent the buffer from growing unchecked.

Red Flags:

• Unbounded write buffer — server accumulates millions of unsent messages for a slow client
• No drop strategy — memory grows until the process is OOM-killed
• Blocking fan-out on a slow client — one slow consumer delays all others in the room
• No client acknowledgment for critical messages — no way to detect or recover from drops

Strategies:

| Strategy | When to Use | Trade-off | |----------|-------------|-----------| | Bounded buffer + drop-oldest | Live telemetry, dashboards | Latest data is kept; historical gaps are acceptable | | Bounded buffer + drop-newest | Order books, audit trails | No overwrite; client must slow down or disconnect | | Client acknowledgment | Payment events, critical notifications | Reliable delivery; adds latency and complexity | | Disconnect slow client | Commodity data streams | Simple; forces client reconnect and re-sync |

Go — bounded channel with drop-oldest:

type BoundedSender struct {
    ch chan []byte
}

func NewBoundedSender(capacity int) *BoundedSender {
    return &BoundedSender{ch: make(chan []byte, capacity)}
}

func (s *BoundedSender) Send(data []byte) {
    select {
    case s.ch  }>();

  track(id: string, payload: unknown, ws: WebSocket, timeoutMs = 5_000): void {
    const timer = setTimeout(() => {
      console.warn('[AckTracker] no ack for', id, '— retrying');
      ws.send(JSON.stringify({ id, payload }));
    }, timeoutMs);
    this.pending.set(id, { payload, timer });
  }

  acknowledge(id: string): void {
    const entry = this.pending.get(id);
    if (entry) { clearTimeout(entry.timer); this.pending.delete(id); }
  }
}

---

9. Real-Time Anti-Patterns

| Anti-Pattern | Description | Fix | |-------------|-------------|-----| | No heartbeat | Idle connections silently dropped by proxies after 60–90 s | Send ping every 25–30 s; reset read deadline on pong | | Unbounded connections | No per-user or total connection cap — file descriptor exhaustion | Enforce maxPerUser in connection manager; set OS ulimit | | Auth in query string | JWT visible in logs, proxy traces, browser history | Use short-lived ticket pattern or `Authoriza

…

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: mickeyyaya
• Source: mickeyyaya/refactoring-skills
• License: MIT

Install and usage instructions live in the source repository linked above.