websockets

name	websockets
description	WebSocket implementation and best practices
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"real-time"}

What I do

• Implement WebSocket connections
• Handle real-time bidirectional communication
• Manage connection lifecycle
• Implement authentication
• Handle reconnection
• Message framing and protocols
• Scale WebSocket connections
• Monitor and debug

When to use me

When implementing real-time features with WebSockets.

WebSocket Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        Client                                    │
│  ┌─────────┐    ┌──────────┐    ┌─────────┐                    │
│  │Socket.io│◄──►│Reconnect │◄──►│Protocol │                    │
│  │ Client  │    │ Logic    │    │Parser   │                    │
│  └─────────┘    └──────────┘    └─────────┘                    │
└────────────────────────────┬────────────────────────────────────┘
                             │ WebSocket
┌────────────────────────────▼────────────────────────────────────┐
│                      Load Balancer                              │
│                    (Sticky Sessions)                           │
└────────────────────────────┬────────────────────────────────────┘
                             │
         ┌───────────────────┼───────────────────┐
         ▼                   ▼                   ▼
┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│ WebSocket   │     │ WebSocket   │     │ WebSocket   │
│ Server 1    │     │ Server 2    │     │ Server 3    │
└─────────────┘     └─────────────┘     └─────────────┘
         │                   │                   │
         └───────────────────┼───────────────────┘
                             │
                    ┌────────▼────────┐
                    │ Message Broker  │
                    │ (Redis Pub/Sub) │
                    └─────────────────┘

Python WebSocket Server

import asyncio
import json
from typing import Dict, Set
from dataclasses import dataclass, field
from enum import Enum


class MessageType(Enum):
    # Client → Server
    SUBSCRIBE = "subscribe"
    UNSUBSCRIBE = "unsubscribe"
    SEND_MESSAGE = "send_message"
    
    # Server → Client
    WELCOME = "welcome"
    MESSAGE = "message"
    ERROR = "error"
    SUBSCRIBED = "subscribed"
    UNSUBSCRIBED = "unsubscribed"


@dataclass
class Client:
    id: str
    websocket: asyncio.WebSocketConnection
    subscriptions: Set[str] = field(default_factory=set)


class WebSocketServer:
    """WebSocket server with subscriptions and broadcasting."""
    
    def __init__(self) -> None:
        self.clients: Dict[str, Client] = {}
        self.subscriptions: Dict[str, Set[str]] = {}  # topic -> client IDs
    
    async  () -> :
        
        client = Client(=client_id, websocket=websocket)
        .clients[client_id] = client
        
        
         .send_message(
            client,
            MessageType.WELCOME.value,
            {: client_id}
        )
        
        :
              message  websocket:
                 .handle_message(client, message)
         asyncio.CancelledError:
            
         Exception  e:
            ()
        :
             .disconnect(client)
    
      () -> :
        
        :
            message = json.loads(raw_message)
            msg_type = message.get()
            payload = message.get(, {})
            
             msg_type == MessageType.SUBSCRIBE.value:
                 .subscribe(client, payload.get())
             msg_type == MessageType.UNSUBSCRIBE.value:
                 .unsubscribe(client, payload.get())
             msg_type == MessageType.SEND_MESSAGE.value:
                 .send_to_topic(
                    payload.get(),
                    payload.get()
                )
            :
                 .send_error(client, )
        
         json.JSONDecodeError:
             .send_error(client, )
    
      () -> :
        
         topic   .subscriptions:
            .subscriptions[topic] = ()
        
        .subscriptions[topic].add(client.)
        client.subscriptions.add(topic)
        
         .send_message(
            client,
            MessageType.SUBSCRIBED.value,
            {: topic}
        )
    
      () -> :
        
         topic  .subscriptions:
            .subscriptions[topic].discard(client.)
        
        client.subscriptions.discard(topic)
        
         .send_message(
            client,
            MessageType.UNSUBSCRIBED.value,
            {: topic}
        )
    
      () -> :
        
         topic   .subscriptions:
            
        
        payload = {
            : MessageType.MESSAGE.value,
            : {
                : topic,
                : message,
            }
        }
        
        
         client_id  .subscriptions[topic]:
             client_id  .clients:
                 .send_message_json(
                    .clients[client_id],
                    payload
                )
    
      () -> :
        
         .send_message_json(client, {
            : msg_type,
            : payload,
        })
    
      () -> :
        
        :
             client.websocket.send(json.dumps(message))
         Exception  e:
            ()
             .disconnect(client)
    
      () -> :
        
         .send_message(
            client,
            MessageType.ERROR.value,
            {: error}
        )
    
      () -> :
        
        
         topic  (client.subscriptions):
             topic  .subscriptions:
                .subscriptions[topic].discard(client.)
        
        
        .clients.pop(client., )
        
        
        :
             client.websocket.close()
         Exception:
            

Socket.IO Implementation

// Server-side (Node.js)
import { Server } from 'socket.io';


const io = new Server(3000, {
  cors: {
    origin: '*',
    methods: ['GET', 'POST'],
  },
  pingTimeout: 60000,
  pingInterval: 25000,
});


// Authentication middleware
io.use((socket, next) => {
  const token = socket.handshake.auth.token;
  
  if (!token) {
    return next(new Error('Authentication required'));
  }
  
  try {
    const user = verifyToken(token);
    socket.user = user;
    next();
  } catch (e) {
    next(new Error('Invalid token'));
  }
});


io.on('connection', (socket) => {
  console.log(`User connected: ${socket.user.id}`);
  
  // Join rooms
  socket.(,  {
    socket.(roomId);
    socket.(roomId).(, {
      : socket..,
      roomId,
    });
  });
  
  
  socket.(,  {
    socket.(roomId);
    socket.(roomId).(, {
      : socket..,
      roomId,
    });
  });
  
  
  socket.(,  {
     { roomId, content } = data;
    
     message = {
      : (),
      content,
      : socket..,
      :  ().(),
    };
    
    
    io.(roomId).(, message);
  });
  
  
  socket.(,  {
    socket.(roomId).(, {
      : socket..,
      : ,
    });
  });
  
  socket.(,  {
    socket.(roomId).(, {
      : socket..,
      : ,
    });
  });
  
  
  socket.(,  {
    .();
    
    
    socket..( {
       (roomId !== socket.) {
        socket.(roomId).(, {
          : socket..,
          roomId,
        });
      }
    });
  });
});

Client Implementation

// Client-side (React)
import { useEffect, useRef, useState, useCallback } from 'react';
import io, { Socket } from 'socket.io-client';


interface UseSocketOptions {
  url: string;
  token: string;
  autoConnect?: boolean;
}


export function useSocket({
  url,
  token,
  autoConnect = true,
}: UseSocketOptions) {
  const socketRef = useRef<Socket | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  const [lastMessage, setLastMessage] = useState<any>(null);
  
  useEffect(() => {
    socketRef.current = io(url, {
      auth: { token },
      transports: ['websocket'],
      reconnection: true,
      reconnectionAttempts: 5,
      reconnectionDelay: 1000,
    });
    
    socketRef.current.on('connect', () => {
      setIsConnected();
      .();
    });
    
    socketRef..(,  {
      ();
      .(, reason);
    });
    
    socketRef..(,  {
      (message);
    });
    
     (autoConnect) {
      socketRef..();
    }
    
      {
      socketRef.?.();
    };
  }, [url, token, autoConnect]);
  
   joinRoom = ( {
    socketRef.?.(, roomId);
  }, []);
  
   leaveRoom = ( {
    socketRef.?.(, roomId);
  }, []);
  
   sendMessage = ( {
    socketRef.?.(, { roomId, content });
  }, []);
  
   on = ( {
    socketRef.?.(event, callback);
      {
      socketRef.?.(event, callback);
    };
  }, []);
  
   {
    isConnected,
    lastMessage,
    joinRoom,
    leaveRoom,
    sendMessage,
    on,
    : socketRef.,
  };
}

Scaling WebSockets

# Using Redis Pub/Sub for horizontal scaling
import aioredis
import json


class RedisPubSubManager:
    """Redis Pub/Sub for multi-instance WebSocket scaling."""
    
    def __init__(self, redis_url: str) -> None:
        self.redis = aioredis.from_url(redis_url)
        self.pubsub = self.redis.pubsub()
        self.subscriptions: Dict[str, Set[str]] = {}
        self.client_messages: asyncio.Queue = asyncio.Queue()
    
    async def subscribe(self, topic: str, server_id: str) -> None:
        """Subscribe to topic across all servers."""
        await self.redis.subscribe(topic)
        
        if topic not in self.subscriptions:
            self.subscriptions[topic] = set()
        
        self.subscriptions[topic].add(server_id)
    
    async def publish(self, topic: str, message: dict) -> None:
        """Publish message to topic."""
        await .redis.publish(
            topic,
            json.dumps({
                : .server_id,
                : message,
            })
        )
    
      () -> :
        
          message  .pubsub.listen():
             message[] == :
                data = json.loads(message[])
                
                
                 data[] != .server_id:
                     .relay_to_local_clients(
                        message[],
                        data[]
                    )

Best Practices

1. Use secure WebSockets (wss://)
   Always use TLS in production

2. Authenticate connections
   Validate tokens on connection

3. Implement heartbeat/ping
   Detect dead connections quickly

4. Handle reconnection
   Exponential backoff with jitter

5. Limit connection lifetime
   Require re-authentication periodically

6. Scale horizontally
   Use Redis Pub/Sub for broadcasting

7. Monitor connections
   Track connection counts, message rates

8. Set message size limits
   Prevent memory exhaustion

9. Handle backpressure
   Don't overwhelm slow clients

10. Graceful shutdown
    Notify clients before disconnecting