realtime-systems

name	realtime-systems
description	WebSocket, real-time communication, and event-driven architectures
domain	development-stacks
version	1.0.0
tags	["websocket","socket-io","sse","pubsub","real-time","events"]
triggers	{"keywords":{"primary":["realtime","real-time","websocket","socket.io","sse","pubsub"],"secondary":["event driven","push","streaming","live","notification","kafka","rabbitmq"]},"context_boost":["chat","collaboration","live update","sync"],"context_penalty":["batch","static","frontend"],"priority":"medium"}

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

Client Implementation

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.();
        ();
      };

      .. =  {
         message = .(event.);
        .(message);
      };

      .. =  {
        .(, event., event.);
        .();
      };

      .. =  {
        .(, error);
        (error);
      };
    });
  }

   () {
     (. >= .) {
      .();
      ;
    }

    .++;
     delay = . * .(, . - );

    .();

    ( {
      .().( {});
    }, delay);
  }

  () {
     (.?. === .) {
      ..(.(message));
    }  {
      
      ..(message);
    }
  }

   () {
     (.. > ) {
       message = ..();
      .(message);
    }
  }

   () {
     handlers = ..(message.);
    handlers?.( (message));

    
     wildcardHandlers = ..();
    wildcardHandlers?.( (message));
  }

  () {
     (!..()) {
      ..(,  ());
    }
    ..()!.(handler);

    
      {
      ..()?.(handler);
    };
  }

  
  () {
    .({ : , room });
  }

  () {
    .({ : , room });
  }

  () {
    .({ : , room, content });
  }

  () {
    .?.();
    . = ;
  }
}


 ws =  ();

ws.(,  {
  .(, msg.);
  ws.();
});

ws.(,  {
  .();
});

 ws.();

---

Socket.IO

Server

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

   {
     user =  (token);
    socket.. = user;
    ();
  }  (err) {
    ( ());
  }
});


 chatNamespace = io.();

chatNamespace.(,  {
   user = socket..;
  .();

  
  socket.();

  
  socket.(,  (: ) => {
    
     hasAccess =  (user., roomId);
     (!hasAccess) {
      socket.(, { :  });
      ;
    }

    socket.(roomId);

    
    socket.(roomId).(, {
      : user.,
      : user.,
    });

    
     messages =  (roomId, );
    socket.(, { roomId, messages });
  });

  
  socket.(,  {
    socket.(roomId);
    socket.(roomId).(, {
      : user.,
      : user.,
    });
  });

  
  socket.(,  (: { : ; :  }) => {
     message = {
      : (),
      : data.,
      : user.,
      : user.,
      : data.,
      :  (),
    };

    
     (message);

    
    chatNamespace.(data.).(, message);
  });

  
  socket.(,  {
    socket.(roomId).(, {
      : user.,
      : user.,
    });
  });

  socket.(,  {
    socket.(roomId).(, {
      : user.,
    });
  });

  
  socket.(,  {
    .();
  });
});


 () {
  chatNamespace.().(event, data);
}

httpServer.();

Client (React)

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;

    const newSocket = io(`${API_URL}/chat`, {
      auth: { token },
      transports: ['websocket'],
    });

    newSocket.on('connect', () => {
      console.log('Socket connected');
    });

    newSocket.on('connect_error', (error) => {
      console.error('Socket connection error:', error);
    });

    setSocket(newSocket);

    return  {
      newSocket.();
    };
  }, [token]);

   (
    
  );
}

  () {
   ();
}


 () {
   socket = ();
   [messages, setMessages] = useState<[]>([]);
   [typingUsers, setTypingUsers] = useState<<>>( ());

  ( {
     (!socket || !roomId) ;

    
    socket.(, roomId);

    
    socket.(,  {
      ( [...prev, message]);
    });

    
    socket.(,  {
      (messages);
    });

    
    socket.(,  {
      (  (prev).(userId));
    });

    socket.(,  {
      ( {
         next =  (prev);
        next.(userId);
         next;
      });
    });

      {
      socket.(, roomId);
      socket.();
      socket.();
      socket.();
      socket.();
    };
  }, [socket, roomId]);

    = () => {
    socket?.(, { roomId, content });
  };

    = () => {
    socket?.(, roomId);
  };

    = () => {
    socket?.(, roomId);
  };

   { messages, typingUsers, sendMessage, startTyping, stopTyping };
}

---

Server-Sent Events (SSE)

Server

import express from 'express';

const app = express();

// SSE endpoint
app.get('/events', (req, res) => {
  // Set SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  // Send initial connection event
  res.write(`event: connected\ndata: ${JSON.stringify({ time: Date.now() })}\n\n`);

  // Keep-alive interval
  const keepAlive = setInterval(() => {
    res.write(`: keep-alive\n\n`);
  }, 30000);

  // Subscribe to events
  const unsubscribe = eventEmitter.on('update', (data) => {
    res.write(`event: update\ndata: ${JSON.stringify(data)}\n\n`);
  });

  // Handle client disconnect
  req.on('close', () => {
    clearInterval(keepAlive);
    unsubscribe();
  });
});


app.(, authenticate,  {
   { userId } = req.;

  res.(, );
  res.(, );
  res.(, );

  
   channel = ;
   unsubscribe = pubsub.(channel,  {
    res.();
  });

  req.(,  {
    ();
  });
});

Client

class EventSourceClient {
  private eventSource: EventSource | null = null;
  private handlers = new Map<string, Set<Function>>();

  connect(url: string) {
    this.eventSource = new EventSource(url);

    this.eventSource.onopen = () => {
      console.log('SSE connected');
    };

    this.eventSource.onerror = (error) => {
      console.error('SSE error:', error);
      // EventSource auto-reconnects
    };

    // Handle named events
    this.handlers.forEach((handlers, eventType) => {
      this.eventSource!.addEventListener(eventType, (event: MessageEvent) => {
        const data = .(event.);
        handlers.( (data));
      });
    });
  }

  () {
     (!..(eventType)) {
      ..(eventType,  ());

      
       (.) {
        ..(eventType,  {
           data = .(event.);
          ..(eventType)?.( (data));
        });
      }
    }

    ..(eventType)!.(handler);

      {
      ..(eventType)?.(handler);
    };
  }

  () {
    .?.();
    . = ;
  }
}


 sse =  ();
sse.(,  .(, data));
sse.(,  (data));
sse.();

---

Pub/Sub with Redis

import Redis from 'ioredis';

const publisher = new Redis(process.env.REDIS_URL);
const subscriber = new Redis(process.env.REDIS_URL);

// Publish event
async function publishEvent(channel: string, event: any) {
  await publisher.publish(channel, JSON.stringify(event));
}

// Subscribe to channel
function subscribe(channel: string, handler: (event: any) => void) {
  subscriber.subscribe(channel);

  subscriber.on('message', (ch, message) => {
    if (ch === channel) {
      handler(JSON.parse(message));
    }
  });
}

// Pattern subscription
function subscribePattern(pattern: string, handler: (channel: , event: ) => ) {
  subscriber.(pattern);

  subscriber.(,  {
     (pat === pattern) {
      (channel, .(message));
    }
  });
}


(,  {
  .(, event);
});

(,  {
   roomId = channel.()[];
  .(, event);
});

(, { : , :  });
(, { : , :  });

---

Related Skills

• [[backend]] - Server implementation
• [[system-design]] - Event-driven architecture
• [[cloud-platforms]] - Managed pub/sub services