realtime-systems

Real-time Systems

Overview

Building real-time applications with WebSocket, Server-Sent Events, and event-driven architectures.

WebSocket

Server Implementation (Node.js)

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);