WebSocket Implementation

Building a WebSocket Server

Let's build real WebSocket servers and clients from scratch. We'll start with the native Node.js ws library, then explore practical patterns for production applications.

Setting Up the Server

// Install: npm install ws

const WebSocket = require('ws');

// Create WebSocket server on port 8080
const wss = new WebSocket.Server({ port: 8080 });

console.log('WebSocket server running on ws://localhost:8080');

// Handle new connections
wss.on('connection', (ws, request) => {
  console.log('New client connected from:', request.socket.remoteAddress);
  
  // Send welcome message
  ws.send(JSON.stringify({ 
    type: 'welcome', 
    message: 'Connected to server!' 
  }));
  
  // Handle incoming messages
  ws.on('message', (data, isBinary) => {
    if (isBinary) {
      console.log('Received binary data:', data.length, 'bytes');
    } else {
      const message = JSON.parse(data.toString());
      console.log('Received:', message);
      handleMessage(ws, message);
    }
  });
  
  // Handle client disconnect
  ws.on('close', (code, reason) => {
    console.log('Client disconnected:', code, reason.toString());
  });
  
  // Handle errors
  ws.on('error', (error) => {
    console.error('WebSocket error:', error);
  });
});

function handleMessage(ws, message) {
  switch (message.type) {
    case 'chat':
      // Broadcast to all clients
      broadcast({ type: 'chat', user: message.user, text: message.text });
      break;
    case 'ping':
      ws.send(JSON.stringify({ type: 'pong', timestamp: Date.now() }));
      break;
  }
}

// Broadcast to all connected clients
function broadcast(message) {
  const data = JSON.stringify(message);
  wss.clients.forEach(client => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(data);
    }
  });
}

Complete Chat Server Example

const WebSocket = require('ws');
const http = require('http');
const { v4: uuidv4 } = require('uuid');

// Create HTTP server (for serving static files or health checks)
const server = http.createServer((req, res) => {
  if (req.url === '/health') {
    res.writeHead(200);
    res.end('OK');
  }
});

// Attach WebSocket to HTTP server
const wss = new WebSocket.Server({ server });

// Store connected clients with metadata
const clients = new Map();

// Store chat rooms
const rooms = new Map();

wss.on('connection', (ws, request) => {
  const clientId = uuidv4();
  const clientInfo = {
    id: clientId,
    ws: ws,
    username: null,
    currentRoom: null,
    joinedAt: new Date()
  };
  clients.set(clientId, clientInfo);
  
  // Send client their ID
  send(ws, { type: 'connected', clientId });
  
  ws.on('message', (data) => {
    try {
      const message = JSON.parse(data.toString());
      handleMessage(clientInfo, message);
    } catch (error) {
      send(ws, { type: 'error', message: 'Invalid message format' });
    }
  });
  
  ws.on('close', () => {
    handleDisconnect(clientInfo);
    clients.delete(clientId);
  });
});

function handleMessage(client, message) {
  switch (message.type) {
    case 'setUsername':
      client.username = message.username;
      send(client.ws, { 
        type: 'usernameSet', 
        username: client.username 
      });
      break;
      
    case 'joinRoom':
      joinRoom(client, message.room);
      break;
      
    case 'leaveRoom':
      leaveRoom(client);
      break;
      
    case 'chatMessage':
      if (client.currentRoom) {
        broadcastToRoom(client.currentRoom, {
          type: 'chatMessage',
          user: client.username,
          text: message.text,
          timestamp: Date.now()
        }, client.id);
      }
      break;
      
    case 'typing':
      if (client.currentRoom) {
        broadcastToRoom(client.currentRoom, {
          type: 'userTyping',
          user: client.username
        }, client.id);
      }
      break;
  }
}

function joinRoom(client, roomName) {
  // Leave current room first
  if (client.currentRoom) {
    leaveRoom(client);
  }
  
  // Create room if doesn't exist
  if (!rooms.has(roomName)) {
    rooms.set(roomName, new Set());
  }
  
  // Join new room
  rooms.get(roomName).add(client.id);
  client.currentRoom = roomName;
  
  // Notify room
  broadcastToRoom(roomName, {
    type: 'userJoined',
    user: client.username,
    room: roomName,
    userCount: rooms.get(roomName).size
  });
  
  // Send room info to client
  send(client.ws, {
    type: 'roomJoined',
    room: roomName,
    userCount: rooms.get(roomName).size
  });
}

function leaveRoom(client) {
  if (!client.currentRoom) return;
  
  const room = rooms.get(client.currentRoom);
  if (room) {
    room.delete(client.id);
    
    broadcastToRoom(client.currentRoom, {
      type: 'userLeft',
      user: client.username,
      userCount: room.size
    });
    
    // Clean up empty rooms
    if (room.size === 0) {
      rooms.delete(client.currentRoom);
    }
  }
  
  client.currentRoom = null;
}

function handleDisconnect(client) {
  if (client.currentRoom) {
    leaveRoom(client);
  }
}

function send(ws, message) {
  if (ws.readyState === WebSocket.OPEN) {
    ws.send(JSON.stringify(message));
  }
}

function broadcastToRoom(roomName, message, excludeClientId = null) {
  const room = rooms.get(roomName);
  if (!room) return;
  
  const data = JSON.stringify(message);
  room.forEach(clientId => {
    if (clientId !== excludeClientId) {
      const client = clients.get(clientId);
      if (client && client.ws.readyState === WebSocket.OPEN) {
        client.ws.send(data);
      }
    }
  });
}

server.listen(8080, () => {
  console.log('Server running on http://localhost:8080');
});

Browser Client Implementation

// chat-client.js - Robust WebSocket client

class ChatClient {
  constructor(url) {
    this.url = url;
    this.ws = null;
    this.clientId = null;
    this.username = null;
    this.currentRoom = null;
    
    // Reconnection settings
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 5;
    this.reconnectDelay = 1000;
    
    // Event handlers (set by application)
    this.onConnected = () => {};
    this.onDisconnected = () => {};
    this.onMessage = () => {};
    this.onError = () => {};
    this.onUserJoined = () => {};
    this.onUserLeft = () => {};
    this.onUserTyping = () => {};
    
    // Message queue for offline sending
    this.messageQueue = [];
  }
  
  connect() {
    return new Promise((resolve, reject) => {
      this.ws = new WebSocket(this.url);
      
      this.ws.onopen = () => {
        console.log('Connected to server');
        this.reconnectAttempts = 0;
        
        // Flush message queue
        this.flushMessageQueue();
        
        resolve();
      };
      
      this.ws.onmessage = (event) => {
        const message = JSON.parse(event.data);
        this.handleMessage(message);
      };
      
      this.ws.onclose = (event) => {
        console.log('Disconnected:', event.code, event.reason);
        this.onDisconnected(event);
        
        // Attempt reconnection for abnormal closures
        if (event.code !== 1000 && event.code !== 1001) {
          this.attemptReconnect();
        }
      };
      
      this.ws.onerror = (error) => {
        console.error('WebSocket error:', error);
        this.onError(error);
        reject(error);
      };
    });
  }
  
  handleMessage(message) {
    switch (message.type) {
      case 'connected':
        this.clientId = message.clientId;
        this.onConnected(message);
        break;
        
      case 'usernameSet':
        this.username = message.username;
        break;
        
      case 'roomJoined':
        this.currentRoom = message.room;
        break;
        
      case 'chatMessage':
        this.onMessage(message);
        break;
        
      case 'userJoined':
        this.onUserJoined(message);
        break;
        
      case 'userLeft':
        this.onUserLeft(message);
        break;
        
      case 'userTyping':
        this.onUserTyping(message);
        break;
        
      case 'error':
        this.onError(new Error(message.message));
        break;
    }
  }
  
  attemptReconnect() {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.log('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(() => {
        // Will retry via onclose handler
      });
    }, delay);
  }
  
  send(message) {
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(message));
    } else {
      // Queue message for when we reconnect
      this.messageQueue.push(message);
    }
  }
  
  flushMessageQueue() {
    while (this.messageQueue.length > 0) {
      const message = this.messageQueue.shift();
      this.send(message);
    }
  }
  
  // Public API methods
  setUsername(username) {
    this.send({ type: 'setUsername', username });
  }
  
  joinRoom(room) {
    this.send({ type: 'joinRoom', room });
  }
  
  leaveRoom() {
    this.send({ type: 'leaveRoom' });
    this.currentRoom = null;
  }
  
  sendMessage(text) {
    this.send({ type: 'chatMessage', text });
  }
  
  sendTyping() {
    this.send({ type: 'typing' });
  }
  
  disconnect() {
    if (this.ws) {
      this.ws.close(1000, 'User disconnected');
    }
  }
}

// Usage
const chat = new ChatClient('wss://api.example.com/chat');

chat.onConnected = () => {
  chat.setUsername('John');
  chat.joinRoom('general');
};

chat.onMessage = (message) => {
  displayMessage(message.user, message.text, message.timestamp);
};

chat.onUserJoined = (event) => {
  displaySystemMessage(`${event.user} joined the room`);
};

chat.onUserTyping = (event) => {
  showTypingIndicator(event.user);
};

await chat.connect();

Binary Data Transfer

// Sending binary data (e.g., file upload over WebSocket)

// Client-side: Sending a file
async function sendFile(ws, file) {
  // Read file as ArrayBuffer
  const buffer = await file.arrayBuffer();
  
  // Send metadata first
  ws.send(JSON.stringify({
    type: 'fileStart',
    name: file.name,
    size: file.size,
    mimeType: file.type
  }));
  
  // Send file in chunks
  const CHUNK_SIZE = 64 * 1024; // 64KB chunks
  const chunks = Math.ceil(buffer.byteLength / CHUNK_SIZE);
  
  for (let i = 0; i < chunks; i++) {
    const start = i * CHUNK_SIZE;
    const end = Math.min(start + CHUNK_SIZE, buffer.byteLength);
    const chunk = buffer.slice(start, end);
    
    // Send chunk index as first 4 bytes, then data
    const message = new ArrayBuffer(4 + chunk.byteLength);
    const view = new DataView(message);
    view.setUint32(0, i);
    new Uint8Array(message, 4).set(new Uint8Array(chunk));
    
    ws.send(message);
    
    // Report progress
    console.log(`Sent chunk ${i + 1}/${chunks}`);
  }
  
  ws.send(JSON.stringify({ type: 'fileEnd' }));
}

// Server-side: Receiving file
const fileBuffers = new Map();

ws.on('message', (data, isBinary) => {
  if (isBinary) {
    // Binary chunk received
    const view = new DataView(data.buffer);
    const chunkIndex = view.getUint32(0);
    const chunkData = data.slice(4);
    
    const clientFile = fileBuffers.get(ws);
    if (clientFile) {
      clientFile.chunks.set(chunkIndex, chunkData);
    }
  } else {
    const message = JSON.parse(data.toString());
    
    if (message.type === 'fileStart') {
      fileBuffers.set(ws, {
        name: message.name,
        size: message.size,
        mimeType: message.mimeType,
        chunks: new Map()
      });
    } else if (message.type === 'fileEnd') {
      const fileData = fileBuffers.get(ws);
      if (fileData) {
        // Reassemble file
        const sortedChunks = [...fileData.chunks.entries()]
          .sort((a, b) => a[0] - b[0])
          .map(([_, chunk]) => chunk);
        
        const completeFile = Buffer.concat(sortedChunks);
        saveFile(fileData.name, completeFile);
        fileBuffers.delete(ws);
      }
    }
  }
});

Heartbeat and Connection Monitoring

// Server-side heartbeat implementation
const WebSocket = require('ws');

const wss = new WebSocket.Server({ port: 8080 });

// Heartbeat interval
const HEARTBEAT_INTERVAL = 30000; // 30 seconds
const CLIENT_TIMEOUT = 35000; // 35 seconds to respond

function heartbeat() {
  this.isAlive = true;
}

wss.on('connection', (ws) => {
  ws.isAlive = true;
  ws.on('pong', heartbeat); // Built-in pong handler
  
  // Also handle application-level ping
  ws.on('message', (data) => {
    const message = JSON.parse(data.toString());
    if (message.type === 'ping') {
      ws.send(JSON.stringify({ 
        type: 'pong', 
        timestamp: message.timestamp,
        serverTime: Date.now()
      }));
    }
  });
});

// Check all connections periodically
const pingInterval = setInterval(() => {
  wss.clients.forEach((ws) => {
    if (ws.isAlive === false) {
      console.log('Terminating dead connection');
      return ws.terminate();
    }
    
    ws.isAlive = false;
    ws.ping(); // Send protocol-level ping
  });
}, HEARTBEAT_INTERVAL);

// Clean up on server close
wss.on('close', () => {
  clearInterval(pingInterval);
});

// Client-side: Monitor connection health
class MonitoredWebSocket {
  constructor(url) {
    this.url = url;
    this.lastPong = Date.now();
    this.connect();
  }
  
  connect() {
    this.ws = new WebSocket(this.url);
    
    this.ws.onopen = () => {
      this.startMonitoring();
    };
    
    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.type === 'pong') {
        this.lastPong = Date.now();
        const latency = this.lastPong - data.timestamp;
        console.log(`Latency: ${latency}ms`);
      }
    };
    
    this.ws.onclose = () => {
      this.stopMonitoring();
    };
  }
  
  startMonitoring() {
    // Send ping every 25 seconds
    this.pingInterval = setInterval(() => {
      if (this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(JSON.stringify({ 
          type: 'ping', 
          timestamp: Date.now() 
        }));
      }
    }, 25000);
    
    // Check for missed pongs
    this.checkInterval = setInterval(() => {
      const timeSinceLastPong = Date.now() - this.lastPong;
      if (timeSinceLastPong > 35000) {
        console.log('Connection appears dead, reconnecting...');
        this.ws.close();
        // Reconnection logic here
      }
    }, 5000);
  }
  
  stopMonitoring() {
    clearInterval(this.pingInterval);
    clearInterval(this.checkInterval);
  }
}

Attaching to Express/HTTP Server

const express = require('express');
const http = require('http');
const WebSocket = require('ws');

const app = express();
const server = http.createServer(app);
const wss = new WebSocket.Server({ server });

// Regular HTTP routes
app.get('/', (req, res) => {
  res.send('Hello HTTP!');
});

app.get('/api/status', (req, res) => {
  res.json({
    connections: wss.clients.size,
    uptime: process.uptime()
  });
});

// WebSocket handling
wss.on('connection', (ws, req) => {
  // Access HTTP request info
  const ip = req.headers['x-forwarded-for'] || req.socket.remoteAddress;
  const userAgent = req.headers['user-agent'];
  
  console.log(`New WS connection from ${ip}`);
  
  ws.on('message', (data) => {
    // Handle messages...
  });
});

// Optional: Different paths for different WebSocket endpoints
const chatWss = new WebSocket.Server({ noServer: true });
const notifyWss = new WebSocket.Server({ noServer: true });

server.on('upgrade', (request, socket, head) => {
  const pathname = new URL(request.url, 'http://localhost').pathname;
  
  if (pathname === '/ws/chat') {
    chatWss.handleUpgrade(request, socket, head, (ws) => {
      chatWss.emit('connection', ws, request);
    });
  } else if (pathname === '/ws/notifications') {
    notifyWss.handleUpgrade(request, socket, head, (ws) => {
      notifyWss.emit('connection', ws, request);
    });
  } else {
    socket.destroy();
  }
});

server.listen(3000, () => {
  console.log('Server running on http://localhost:3000');
});