realtime-systems

# Real-time Systems ## Overview Building real-time applications with WebSocket, Server-Sent Events, and event-driven architectures. --- ## WebSocket ### Server Implementation (Node.js) ```typescript import { WebSocketServer, WebSocket } from 'ws'; import { createServer } from 'http'; import { v4 as uuid } from 'uuid'; const server = createServer(); const wss = new WebSocketServer({ server }); interface Client { id: string; ws: WebSocket; userId?: string; rooms: Set<string>; } const clients = new Map<string, Client>(); const rooms = new Map<string, Set<string>>(); wss.on('connection', (ws, req) => { const clientId = uuid(); const client: Client = { id: clientId, ws, rooms: new Set(), }; clients.set(clientId, client); console.log(`Client connected: ${clientId}`); // Handle messages ws.on('message', (data) => { try { const message = JSON.parse(data.toString()); handleMessage(client, message); } catch (error) { console.error('Invalid message:', error); } }); // Handle disconnection ws.on('close', () => { // Leave all rooms client.rooms.forEach(room => leaveRoom(client, room)); clients.delete(clientId); console.log(`Client disconnected: ${clientId}`); }); // Send connection confirmation send(ws, { type: 'connected', clientId }); }); function handleMessage(client: Client, message: any) { switch (message.type) { case 'authenticate': client.userId = message.userId; break; case 'join': joinRoom(client, message.room); break; case 'leave': leaveRoom(client, message.room); break; case 'message': broadcastToRoom(message.room, { type: 'message', from: client.userId, content: message.content, timestamp: Date.now(), }, client.id); break; case 'ping': send(client.ws, { type: 'pong' }); break; } } function joinRoom(client: Client, room: string) { if (!rooms.has(room)) { rooms.set(room, new Set()); } rooms.get(room)!.add(client.id); client.rooms.add(room); // Notify room members broadcastToRoom(room, { type: 'user_joined', userId: client.userId, room, }, client.id); } function leaveRoom(client: Client, room: string) { rooms.get(room)?.delete(client.id); client.rooms.delete(room); // Notify room members broadcastToRoom(room, { type: 'user_left', userId: client.userId, room, }); } function broadcastToRoom(room: string, message: any, excludeClientId?: string) { const roomClients = rooms.get(room); if (!roomClients) return; roomClients.forEach(clientId => { if (clientId !== excludeClientId) { const client = clients.get(clientId); if (client?.ws.readyState === WebSocket.OPEN) { send(client.ws, message); } } }); } function send(ws: WebSocket, message: any) { ws.send(JSON.stringify(message)); } server.listen(8080); ``` ### Client Implementation ```typescript class WebSocketClient { private ws: WebSocket | null = null; private reconnectAttempts = 0; private maxReconnectAttempts = 5; private reconnectDelay = 1000; private messageHandlers = new Map<string, Set<Function>>(); private messageQueue: any[] = []; constructor(private url: string) {} connect(): Promise<void> { return new Promise((resolve, reject) => { this.ws = new WebSocket(this.url); this.ws.onopen = () => { console.log('WebSocket connected'); this.reconnectAttempts = 0; this.flushMessageQueue(); resolve(); }; this.ws.onmessage = (event) => { const message = JSON.parse(event.data); this.handleMessage(message); }; this.ws.onclose = (event) => { console.log('WebSocket closed:', event.code, event.reason); this.attemptReconnect(); }; this.ws.onerror = (error) => { console.error('WebSocket error:', error); reject(error); }; }); } private attemptReconnect() { if (this.reconnectAttempts >= this.maxReconnectAttempts) { console.error('Max reconnection attempts reached'); return; } this.reconnectAttempts++; const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1); console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`); setTimeout(() => { this.connect().catch(() => {}); }, delay); } send(message: any) { if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(JSON.stringify(message)); } else { // Queue message for when connection is restored this.messageQueue.push(message); } } private flushMessageQueue() { while (this.messageQueue.length > 0) { const message = this.messageQueue.shift(); this.send(message); } } private handleMessage(message: any) { const handlers = this.messageHandlers.get(message.type); handlers?.forEach(handler => handler(message)); // Also emit to wildcard handlers const wildcardHandlers = this.messageHandlers.get('*'); wildcardHandlers?.forEach(handler => handler(message)); } on(type: string, handler: Function) { if (!this.messageHandlers.has(type)) { this.messageHandlers.set(type, new Set()); } this.messageHandlers.get(type)!.add(handler); // Return unsubscribe function return () => { this.messageHandlers.get(type)?.delete(handler); }; } // Convenience methods joinRoom(room: string) { this.send({ type: 'join', room }); } leaveRoom(room: string) { this.send({ type: 'leave', room }); } sendMessage(room: string, content: string) { this.send({ type: 'message', room, content }); } disconnect() { this.ws?.close(); this.ws = null; } } // Usage const ws = new WebSocketClient('wss://api.example.com/ws'); ws.on('connected', (msg) => { console.log('Connected with ID:', msg.clientId); ws.joinRoom('general'); }); ws.on('message', (msg) => { console.log(`[${msg.from}]: ${msg.content}`); }); await ws.connect(); ``` --- ## Socket.IO ### Server ```typescript import { Server } from 'socket.io'; import { createServer } from 'http'; import { createAdapter } from '@socket.io/redis-adapter'; import { createClient } from 'redis'; const httpServer = createServer(); const io = new Server(httpServer, { cors: { origin: process.env.ALLOWED_ORIGINS?.split(',') || '*', credentials: true, }, }); // Redis adapter for horizontal scaling const pubClient = createClient({ url: process.env.REDIS_URL }); const subClient = pubClient.duplicate(); Promise.all([pubClient.connect(), subClient.connect()]).then(() => { io.adapter(createAdapter(pubClient, subClient)); }); // Authentication middleware io.use(async (socket, next) => { const token = socket.handshake.auth.token; try { const user = await verifyToken(token); socket.data.user = user; next(); } catch (err) { next(new Error('Authentication failed')); } }); // Namespace for chat const chatNamespace = io.of('/chat'); chatNamespace.on('connection', (socket) => { const user = socket.data.user; console.log(`User connected: ${user.name}`); // Join user's personal room socket.join(`user:${user.id}`); // Join a chat room socket.on('join_room', async (roomId: string) => { // Verify access const hasAccess = await checkRoomAccess(user.id, roomId); if (!hasAccess) { socket.emit('error', { message: 'Access denied' }); return; } socket.join(roomId); // Notify room members socket.to(roomId).emit('user_joined', { userId: user.id, userName: user.name, }); // Send recent messages const messages = await getRecentMessages(roomId, 50); socket.emit('room_history', { roomId, messages }); }); // Leave room socket.on('leave_room', (roomId: string) => { socket.leave(roomId); socket.to(roomId).emit('user_left', { userId: user.id, userName: user.name, }); }); // Send message socket.on('message', async (data: { roomId: string; content: string }) => { const message = { id: uuid(), roomId: data.roomId, userId: user.id, userName: user.name, content: data.content, timestamp: new Date(), }; // Persist message await saveMessage(message); // Broadcast to room chatNamespace.to(data.roomId).emit('message', message); }); // Typing indicator socket.on('typing_start', (roomId: string) => { socket.to(roomId).emit('user_typing', { userId: user.id, userName: user.name, }); }); socket.on('typing_stop', (roomId: string) => { socket.to(roomId).emit('user_stopped_typing', { userId: user.id, }); }); // Disconnect socket.on('disconnect', () => { console.log(`User disconnected: ${user.name}`); }); }); // Send to specific user (from anywhere in the app) function sendToUser(userId: string, event: string, data: any) { chatNamespace.to(`user:${userId}`).emit(event, data); } httpServer.listen(3000); ``` ### Client (React) ```tsx import { io, Socket } from 'socket.io-client'; import { createContext, useContext, useEffect, useState } from 'react'; // Socket context const SocketContext = createContext<Socket | null>(null); export function SocketProvider({ children }: { children: React.ReactNode }) { const [socket, setSocket] = useState<Socket | null>(null); const { token } = useAuth(); useEffect(() => { if (!token) return;