websocket-implementation

name	websocket-implementation
description	Implement real-time bidirectional communication with WebSockets including connection management, message routing, and scaling. Use when building real-time features, chat systems, live notifications, or collaborative applications.

WebSocket Implementation

Overview

Build scalable WebSocket systems for real-time communication with proper connection management, message routing, error handling, and horizontal scaling support.

When to Use

• Building real-time chat and messaging
• Implementing live notifications
• Creating collaborative editing tools
• Broadcasting live data updates
• Building real-time dashboards
• Streaming events to clients
• Live multiplayer games

Instructions

1. Node.js WebSocket Server (Socket.IO)

const express = require('express');
const http = require('http');
const socketIo = require('socket.io');
const redis = require('redis');

const app = express();
const server = http.createServer(app);
const io = socketIo(server, {
  cors: { origin: '*' },
  transports: ['websocket', 'polling'],
  reconnection: true,
  reconnectionDelay: 1000,
  reconnectionDelayMax: 5000,
  reconnectionAttempts: 5
});

// Redis adapter for horizontal scaling
const redisClient = redis.createClient();
const { createAdapter } = require('@socket.io/redis-adapter');

io.adapter(createAdapter(redisClient, redisClient.duplicate()));

// Connection management
const connectedUsers = new Map();

io.on('connection', (socket) => {
  console.log(`User connected: ${socket.id}`);

  // Store user connection
  socket.on('auth', (userData) => {
    connectedUsers.set(socket.id, {
      userId: userData.id,
      username: userData.username,
      socketId: socket.id,
      connectedAt: new Date()
    });

    // Join user-specific room
    socket.join(`user:${userData.id}`);
    socket.join('authenticated_users');

    // Notify others user is online
    io.to('authenticated_users').emit('user:online', {
      userId: userData.id,
      username: userData.username,
      timestamp: new Date()
    });

    console.log(`User authenticated: ${userData.username}`);
  });

  // Chat messaging
  socket.on('chat:message', (message) => {
    const user = connectedUsers.get(socket.id);

    if (!user) {
      socket.emit('error', { message: 'Not authenticated' });
      return;
    }

    const chatMessage = {
      id: `msg_${Date.now()}`,
      senderId: user.userId,
      senderName: user.username,
      text: message.text,
      roomId: message.roomId,
      timestamp: new Date(),
      status: 'delivered'
    };

    // Save to database
    Message.create(chatMessage);

    // Broadcast to room
    io.to(`room:${message.roomId}`).emit('chat:message', chatMessage);

    // Update message status
    setTimeout(() => {
      socket.emit('chat:message:ack', { messageId: chatMessage.id, status: 'read' });
    }, 100);
  });

  // Room management
  socket.on('room:join', (roomId) => {
    socket.join(`room:${roomId}`);

    const user = connectedUsers.get(socket.id);
    io.to(`room:${roomId}`).emit('room:user:joined', {
      userId: user.userId,
      username: user.username,
      timestamp: new Date()
    });
  });

  socket.on('room:leave', (roomId) => {
    socket.leave(`room:${roomId}`);

    const user = connectedUsers.get(socket.id);
    io.to(`room:${roomId}`).emit('room:user:left', {
      userId: user.userId,
      timestamp: new Date()
    });
  });

  // Typing indicator
  socket.on('typing:start', (roomId) => {
    const user = connectedUsers.get(socket.id);
    io.to(`room:${roomId}`).emit('typing:indicator', {
      userId: user.userId,
      username: user.username,
      isTyping: true
    });
  });

  socket.on('typing:stop', (roomId) => {
    const user = connectedUsers.get(socket.id);
    io.to(`room:${roomId}`).emit('typing:indicator', {
      userId: user.userId,
      isTyping: false
    });
  });

  // Handle disconnection
  socket.on('disconnect', () => {
    const user = connectedUsers.get(socket.id);

    if (user) {
      connectedUsers.delete(socket.id);
      io.to('authenticated_users').emit('user:offline', {
        userId: user.userId,
        timestamp: new Date()
      });

      console.log(`User disconnected: ${user.username}`);
    }
  });

  // Error handling
  socket.on('error', (error) => {
    console.error(`Socket error: ${error}`);
    socket.emit('error', { message: 'An error occurred' });
  });
});

// Server methods
const broadcastUserUpdate = (userId, data) => {
  io.to(`user:${userId}`).emit('user:update', data);
};

const notifyRoom = (roomId, event, data) => {
  io.to(`room:${roomId}`).emit(event, data);
};

const sendDirectMessage = (userId, event, data) => {
  io.to(`user:${userId}`).emit(event, data);
};

server.listen(3000, () => {
  console.log('WebSocket server listening on port 3000');
});

2. Browser WebSocket Client

class WebSocketClient {
  constructor(url, options = {}) {
    this.url = url;
    this.socket = null;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = options.maxReconnectAttempts || 5;
    this.reconnectDelay = options.reconnectDelay || 1000;
    this.listeners = new Map();
    this.messageQueue = [];
    this.isAuthenticated = false;

    this.connect();
  }

  connect() {
    this.socket = io(this.url, {
      reconnection: true,
      reconnectionDelay: this.reconnectDelay,
      reconnectionAttempts: this.maxReconnectAttempts
    });

    this.socket.on('connect',  {
      .();
      . = ;
      .();
    });

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

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

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

  () {
    ..(, userData,  {
       (response.) {
        . = ;
        .();
      }
    });
  }

  () {
    ..(event, callback);

     (!..(event)) {
      ..(event, []);
    }
    ..(event).(callback);
  }

  () {
     (!..) {
      ..({ event, data, callback });
      ;
    }

    ..(event, data, callback);
  }

  () {
     (.. > ) {
       { event, data, callback } = ..();
      ..(event, data, callback);
    }
  }

  () {
    .(, roomId);
  }

  () {
    .(, roomId);
  }

  () {
    .(, { roomId, text });
  }

  () {
     (isTyping) {
      .(, roomId);
    }  {
      .(, roomId);
    }
  }

  () {
    ..();
  }
}


 client =  ();

client.(,  {
  .(, message);
  (message);
});

client.(,  {
  (data);
});

client.(,  {
  (user., );
});

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

3. Python WebSocket Server (aiohttp)

from aiohttp import web
import aiohttp
import json
from datetime import datetime
from typing import Set

class WebSocketServer:
    def __init__(self):
        self.app = web.Application()
        self.rooms = {}
        self.users = {}
        self.setup_routes()

    def setup_routes(self):
        self.app.router.add_get('/ws', self.websocket_handler)
        self.app.router.add_post('/api/message', self.send_message_api)

    async def websocket_handler(self, request):
        ws = web.WebSocketResponse()
        await ws.prepare(request)

        user_id = None
        room_id = None

        async for msg in ws.iter_any():
            if isinstance(msg, aiohttp.WSMessage):
                data = json.loads(msg.data)
                event_type = data.get('type')

                try:
                    if event_type == 'auth':
                        user_id = data.get('userId')
                        self.users[user_id] = ws
                         ws.send_json({
                            : ,
                            : datetime.now().isoformat()
                        })

                     event_type == :
                        room_id = data.get()
                         room_id   .rooms:
                            .rooms[room_id] = ()
                        .rooms[room_id].add(user_id)

                        
                         .broadcast_to_room(room_id, {
                            : ,
                            : user_id,
                            : datetime.now().isoformat()
                        }, exclude=user_id)

                     event_type == :
                        message = {
                            : ,
                            : user_id,
                            : data.get(),
                            : room_id,
                            : datetime.now().isoformat()
                        }

                        
                         .save_message(message)

                        
                         .broadcast_to_room(room_id, message)

                     event_type == :
                         room_id  .rooms:
                            .rooms[room_id].discard(user_id)

                 Exception  error:
                     ws.send_json({
                        : ,
                        : (error)
                    })

        
         user_id:
             .users[user_id]
         room_id  user_id:
             room_id  .rooms:
                .rooms[room_id].discard(user_id)

         ws

      ():
         room_id   .rooms:
            

         user_id  .rooms[room_id]:
             user_id != exclude  user_id  .users:
                :
                     .users[user_id].send_json(message)
                 Exception  error:
                    ()

      ():
        
        

      ():
        data =  request.json()
        room_id = data.get()

         .broadcast_to_room(room_id, {
            : ,
            : data.get(),
            : datetime.now().isoformat()
        })

         web.json_response({: })

 ():
    server = WebSocketServer()
     server.app

 __name__ == :
    app = create_app()
    web.run_app(app, port=)

4. Message Types and Protocols

// Authentication
{
  "type": "auth",
  "userId": "user123",
  "token": "jwt_token_here"
}

// Chat Message
{
  "type": "message",
  "roomId": "room123",
  "text": "Hello everyone!",
  "timestamp": "2025-01-15T10:30:00Z"
}

// Typing Indicator
{
  "type": "typing",
  "roomId": "room123",
  "isTyping": true
}

// Presence
{
  "type": "presence",
  "status": "online|away|offline"
}

// Notification

   
   
   
   

5. Scaling with Redis

const redis = require('redis');
const { createAdapter } = require('@socket.io/redis-adapter');
const { createClient } = require('redis');

const pubClient = createClient({ host: 'redis', port: 6379 });
const subClient = pubClient.duplicate();

io.adapter(createAdapter(pubClient, subClient));

// Publish to multiple servers
io.emit('user:action', { userId: 123, action: 'login' });

// Subscribe to events from other servers
redisClient.subscribe('notifications', (message) => {
  const notification = JSON.parse(message);
  io.to(`user:${notification.userId}`).emit('notification', notification);
});

Best Practices

✅ DO

• Implement proper authentication
• Handle reconnection gracefully
• Manage rooms/channels effectively
• Persist messages appropriately
• Monitor active connections
• Implement presence features
• Use Redis for scaling
• Add message acknowledgment
• Implement rate limiting
• Handle errors properly

❌ DON'T

• Send unencrypted sensitive data
• Keep unlimited message history in memory
• Allow arbitrary room/channel creation
• Forget to clean up disconnected connections
• Send large messages frequently
• Ignore network failures
• Store passwords in messages
• Skip authentication/authorization
• Create unbounded growth of connections
• Ignore scalability from day one

Monitoring

// Track active connections
io.engine.on('connection_error', (err) => {
  console.log(err.req); // the request object
  console.log(err.code); // the error code, e.g. 1
  console.log(err.message); // the error message
  console.log(err.context); // some additional error context
});

app.get('/metrics/websocket', (req, res) => {
  res.json({
    activeConnections: io.engine.clientsCount,
    connectedSockets: io.sockets.sockets.size,
    rooms: Object.keys(io.sockets.adapter.rooms)
  });
});