WebSocket & Real-Time Communication

SSE and WebSocket patterns for real-time data streaming.

When to Apply

• Chat streaming (SSE)
• Live market price updates
• Real-time notifications
• WebSocket connections for bidirectional communication

1. SSE Streaming (CRITICAL - our chat uses this)

FastAPI backend

from fastapi.responses import StreamingResponse
import json
import asyncio

@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def event_generator():
        try:
            async for chunk in process_chat(request):
                yield f"data: {json.dumps(chunk)}\n\n"
            yield "data: [DONE]\n\n"
        except asyncio.CancelledError:
            logger.info("Client disconnected")
        except Exception as e:
            logger.error(f"Stream error: {e}")
            yield f"data: {json.dumps({'error': str(e)})}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Nginx: disable buffering
        }
    )

Frontend client

async function streamChat(message, onChunk, onDone, onError) {
  const controller = new AbortController();
  
  try {
    const response = await fetch('/api/chat/stream', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ message }),
      signal: controller.signal
    });
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';
    
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop(); // Keep incomplete line
      
      for (const line of lines) {
        if (!line.startsWith('data: ')) continue;
        const data = line.slice(6);
        if (data === '[DONE]') { onDone(); return; }
        
        try {
          const chunk = JSON.parse(data);
          if (chunk.error) { onError(chunk.error); return; }
          onChunk(chunk);
        } catch (e) { /* skip malformed */ }
      }
    }
    onDone();
  } catch (e) {
    if (e.name !== 'AbortError') onError(e.message);
  }
  
  return () => controller.abort(); // Cleanup function
}

2. EventSource with Reconnection (HIGH)

class ReconnectingEventSource {
  constructor(url, options = {}) {
    this.url = url;
    this.maxRetries = options.maxRetries || 10;
    this.retryCount = 0;
    this.handlers = {};
    this.connect();
  }
  
  connect() {
    this.source = new EventSource(this.url);
    
    this.source.onopen = () => {
      this.retryCount = 0;
      this.handlers.open?.();
    };
    
    this.source.onmessage = (e) => {
      this.handlers.message?.(JSON.parse(e.data));
    };
    
    this.source.onerror = () => {
      this.source.close();
      if (this.retryCount < this.maxRetries) {
        const delay = Math.min(1000 * 2 ** this.retryCount, 30000);
        this.retryCount++;
        setTimeout(() => this.connect(), delay);
      }
    };
  }
  
  on(event, handler) { this.handlers[event] = handler; }
  close() { this.source?.close(); }
}

3. Heartbeat / Keep-Alive (HIGH)

async def event_generator_with_heartbeat(data_source):
    """SSE generator with periodic heartbeat."""
    last_data = time.time()
    
    while True:
        try:
            data = await asyncio.wait_for(
                data_source.get_next(),
                timeout=15.0
            )
            yield f"data: {json.dumps(data)}\n\n"
            last_data = time.time()
        except asyncio.TimeoutError:
            # Send heartbeat comment (SSE standard)
            yield ":heartbeat\n\n"

4. Nginx SSE Configuration (CRITICAL)

location /api/chat/ {
    proxy_pass http://localhost:8000;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    
    # SSE support - MUST disable buffering
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 300s;
    proxy_connect_timeout 10s;
    chunked_transfer_encoding on;
    
    # Headers
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}

5. Anti-Patterns

• Not authenticating connections during handshake
• Missing heartbeat (stale connections pile up)
• Using WebSocket when SSE suffices (server-to-client only)
• No message size limits
• Blocking event loop with sync processing
• No reconnection logic on client
• Nginx buffering SSE responses (breaks streaming)

6. Checklist

• SSE: X-Accel-Buffering: no header set
• Nginx: proxy_buffering off for SSE routes
• Client implements exponential backoff reconnection
• Heartbeat keeps connections alive
• Cleanup on component unmount (AbortController)
• Error events sent to client on failures
• SSE used for server→client (not WebSocket)