WebSockets & Real-Time Communication

1// ============================================
2// WebSocket & Real-Time Communication Patterns
3// ============================================
4
5// ---------- Pattern 1: Simple WebSocket Server ----------
6
7const WebSocket = require('ws');
8const http = require('http');
9
10const server = http.createServer();
11const wss = new WebSocket.Server({ server });
12
13// Track all connected clients
14const clients = new Map(); // userId → WebSocket connection
15
16wss.on('connection', (ws, req) => {
17  const userId = req.url.split('?userId=')[1];
18  clients.set(userId, ws);
19  console.log(`User \${userId} connected. Total: \${clients.size}`);
20
21  ws.on('message', (data) => {
22    const message = JSON.parse(data);
23    handleMessage(userId, message);
24  });
25
26  ws.on('close', () => {
27    clients.delete(userId);
28    console.log(`User \${userId} disconnected. Total: \${clients.size}`);
29  });
30
31  // Send heartbeat every 30 seconds to detect dead connections
32  const heartbeat = setInterval(() => {
33    if (ws.readyState === WebSocket.OPEN) {
34      ws.ping();
35    } else {
36      clearInterval(heartbeat);
37    }
38  }, 30000);
39});
40
41function handleMessage(senderId, message) {
42  switch (message.type) {
43    case 'direct_message':
44      // Send to specific user
45      const recipientWs = clients.get(message.recipientId);
46      if (recipientWs && recipientWs.readyState === WebSocket.OPEN) {
47        recipientWs.send(JSON.stringify({
48          type: 'new_message',
49          from: senderId,
50          content: message.content,
51          timestamp: Date.now(),
52        }));
53      }
54      break;
55
56    case 'broadcast':
57      // Send to all connected users
58      clients.forEach((ws, id) => {
59        if (id !== senderId && ws.readyState === WebSocket.OPEN) {
60          ws.send(JSON.stringify({
61            type: 'broadcast',
62            from: senderId,
63            content: message.content,
64          }));
65        }
66      });
67      break;
68  }
69}
70
71// ---------- Pattern 2: Scaling with Pub/Sub (Redis) ----------
72
73// ❌ BAD: Single-server WebSocket — doesn't scale
74// When you have 2+ servers behind a load balancer,
75// User A on Server 1 can't send to User B on Server 2
76
77// ✅ GOOD: Redis Pub/Sub for cross-server messaging
78const Redis = require('ioredis');
79const pub = new Redis();  // Publisher
80const sub = new Redis();  // Subscriber
81
82// Each server subscribes to a channel
83sub.subscribe('chat-messages');
84
85sub.on('message', (channel, data) => {
86  const message = JSON.parse(data);
87  // Deliver to local clients connected to THIS server
88  const recipientWs = clients.get(message.recipientId);
89  if (recipientWs && recipientWs.readyState === WebSocket.OPEN) {
90    recipientWs.send(JSON.stringify(message));
91  }
92});
93
94// When a user sends a message, publish to Redis (all servers receive it)
95function handleMessageScalable(senderId, message) {
96  pub.publish('chat-messages', JSON.stringify({
97    type: 'new_message',
98    from: senderId,
99    recipientId: message.recipientId,
100    content: message.content,
101    timestamp: Date.now(),
102    sourceServer: process.env.SERVER_ID, // Track which server sent it
103  }));
104}
105
106// ---------- Pattern 3: Server-Sent Events (SSE) — Simpler Alternative ----------
107
108// SSE is one-way (server → client), uses standard HTTP, auto-reconnects
109const express = require('express');
110const sseApp = express();
111
112sseApp.get('/api/events', (req, res) => {
113  // Set SSE headers
114  res.set({
115    'Content-Type': 'text/event-stream',
116    'Cache-Control': 'no-cache',
117    'Connection': 'keep-alive',
118  });
119
120  // Send a comment to prevent proxy timeout
121  res.write(':keep-alive\n\n');
122
123  // Send events as they happen
124  const sendEvent = (eventType, data) => {
125    res.write(`event: \${eventType}\n`);
126    res.write(`data: \${JSON.stringify(data)}\n\n`);
127  };
128
129  // Example: Send stock price updates every second
130  const interval = setInterval(() => {
131    sendEvent('price-update', {
132      symbol: 'AAPL',
133      price: (150 + Math.random() * 10).toFixed(2),
134      timestamp: Date.now(),
135    });
136  }, 1000);
137
138  // Cleanup on disconnect
139  req.on('close', () => {
140    clearInterval(interval);
141    console.log('SSE client disconnected');
142  });
143});
144
145// Client-side SSE (in browser):
146// const eventSource = new EventSource('/api/events');
147// eventSource.addEventListener('price-update', (event) => {
148//   const data = JSON.parse(event.data);
149//   updateUI(data);
150// });
151
152server.listen(8080, () => console.log('WebSocket server on port 8080'));