WebSocket Programming: Real-Time bidirectional Communication

from enum import IntEnum
from dataclasses import dataclass

class Opcode(IntEnum):
    CONTINUATION = 0x0
    TEXT = 0x1
    BINARY = 0x2
    CLOSE = 0x8
    PING = 0x9
    PONG = 0xA

class WebSocketFrame:
    def __init__(
        self,
        opcode: Opcode,
        payload: bytes,
        fin: bool = True,
        rsv1: bool = False,
        rsv2: bool = False,
        rsv3: bool = False,
        masked: bool = False,
        mask_key: bytes = None
    ):
        self.opcode = opcode
        self.payload = payload
        self.fin = fin
        self.rsv1 = rsv1
        self.rsv2 = rsv2
        self.rsv3 = rsv3
        self.masked = masked
        self.mask_key = mask_key or b''
    
    def to_bytes(self) -> bytes:
        first_byte = (0x80 if self.fin else 0) | self.opcode
        
        if self.payload_length < 126:
            second_byte = (0x80 if self.masked else 0) | self.payload_length
            length_byte = second_byte.to_bytes(1, 'big')
        elif self.payload_length < 65536:
            second_byte = (0x80 if self.masked else 0) | 126
            length_bytes = second_byte.to_bytes(1, 'big') + \
                          self.payload_length.to_bytes(2, 'big')
        else:
            second_byte = (0x80 if self.masked else 0) | 127
            length_bytes = second_byte.to_bytes(1, 'big') + \
                          self.payload_length.to_bytes(8, 'big')
        
        if self.masked and self.mask_key:
            masked_payload = self._mask_payload(self.payload)
            return bytes([first_byte]) + length_bytes + self.mask_key + masked_payload
        
        return bytes([first_byte]) + length_bytes + self.payload
    
    @property
    def payload_length(self) -> int:
        return len(self.payload)
    
    def _mask_payload(self, payload: bytes) -> bytes:
        return bytes(b ^ self.mask_key[i % 4] for i, b in enumerate(payload))

import asyncio
import websockets
import json
from typing import Set, Dict, Any
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class WebSocketServer:
    def __init__(self, host: str = "localhost", port: int = 8765):
        self.host = host
        self.port = port
        self.clients: Set[websockets.WebSocketServerProtocol] = set()
        self.rooms: Dict[str, Set[websockets.WebSocketServerProtocol]] = {}
        self.message_handlers: Dict[str, callable] = {}
    
    async def register(self, websocket: websockets.WebSocketServerProtocol):
        self.clients.add(websocket)
        logger.info(f"Client connected: {websocket.remote_address}")
    
    async def unregister(self, websocket: websockets.WebSocketServerProtocol):
        self.clients.discard(websocket)
        
        for room in list(self.rooms.values()):
            room.discard(websocket)
        
        logger.info(f"Client disconnected: {websocket.remote_address}")
    
    async def handle_message(self, websocket: websockets.WebSocketServerProtocol, message: str):
        try:
            data = json.loads(message)
            msg_type = data.get('type', 'unknown')
            
            if msg_type in self.message_handlers:
                await self.message_handlers[msg_type](websocket, data)
            else:
                logger.warning(f"Unknown message type: {msg_type}")
        
        except json.JSONDecodeError:
            logger.error(f"Invalid JSON: {message}")
        except Exception as e:
            logger.error(f"Error handling message: {e}")
    
    async def broadcast(self, message: Dict[str, Any], exclude: Set = None):
        exclude = exclude or set()
        message_str = json.dumps(message)
        
        await asyncio.gather(
            *[
                client.send(message_str)
                for client in self.clients
                if client not in exclude and client.open
            ],
            return_exceptions=True
        )
    
    async def send_to_room(self, room: str, message: Dict[str, Any]):
        if room in self.rooms:
            message_str = json.dumps(message)
            await asyncio.gather(
                *[
                    client.send(message_str)
                    for client in self.rooms[room]
                    if client.open
                ],
                return_exceptions=True
            )
    
    async def join_room(self, websocket: websockets.WebSocketServerProtocol, room: str):
        if room not in self.rooms:
            self.rooms[room] = set()
        self.rooms[room].add(websocket)
        await websocket.send(json.dumps({
            'type': 'room_joined',
            'room': room
        }))
    
    async def leave_room(self, websocket: websockets.WebSocketServerProtocol, room: str):
        if room in self.rooms:
            self.rooms[room].discard(websocket)
    
    async def handler(self, websocket: websockets.WebSocketServerProtocol):
        await self.register(websocket)
        
        try:
            async for message in websocket:
                await self.handle_message(websocket, message)
        
        except websockets.exceptions.ConnectionClosed:
            pass
        finally:
            await self.unregister(websocket)
    
    async def start(self):
        async with websockets.serve(self.handler, self.host, self.port):
            logger.info(f"WebSocket server started on {self.host}:{self.port}")
            await asyncio.Future()
    
    def register_handler(self, msg_type: str, handler: callable):
        self.message_handlers[msg_type] = handler

class ChatMessageHandler:
    def __init__(self, server: WebSocketServer):
        self.server = server
        self._register_handlers()
    
    def _register_handlers(self):
        self.server.register_handler('chat_message', self.handle_chat)
        self.server.register_handler('join_room', self.handle_join_room)
        self.server.register_handler('leave_room', self.handle_leave_room)
        self.server.register_handler('typing', self.handle_typing)
    
    async def handle_chat(self, websocket, data: Dict):
        room = data.get('room')
        message = data.get('message')
        username = data.get('username', 'Anonymous')
        
        await self.server.send_to_room(room, {
            'type': 'chat_message',
            'username': username,
            'message': message,
            'timestamp': datetime.utcnow().isoformat()
        })
    
    async def handle_join_room(self, websocket, data: Dict):
        room = data.get('room')
        await self.server.join_room(websocket, room)
        
        await self.server.send_to_room(room, {
            'type': 'user_joined',
            'room': room,
            'timestamp': datetime.utcnow().isoformat()
        })
    
    async def handle_leave_room(self, websocket, data: Dict):
        room = data.get('room')
        await self.server.leave_room(websocket, room)
    
    async def handle_typing(self, websocket, data: Dict):
        room = data.get('room')
        username = data.get('username')
        
        await self.server.send_to_room(room, {
            'type': 'typing',
            'username': username
        })

const WebSocket = require('ws');

class WebSocketServer {
    constructor(port) {
        this.port = port;
        this.wss = new WebSocket.Server({ port });
        this.clients = new Map();
        this.rooms = new Map();
        
        this.wss.on('connection', this.handleConnection.bind(this));
    }
    
    handleConnection(ws, req) {
        const clientId = this.generateClientId();
        this.clients.set(clientId, { ws, rooms: new Set() });
        
        console.log(`Client connected: ${clientId}`);
        
        ws.on('message', (message) => {
            this.handleMessage(clientId, message);
        });
        
        ws.on('close', () => {
            this.handleDisconnect(clientId);
        });
        
        ws.on('error', (error) => {
            console.error(`WebSocket error: ${error}`);
        });
        
        this.send(clientId, { type: 'connected', clientId });
    }
    
    handleMessage(clientId, rawMessage) {
        try {
            const message = JSON.parse(rawMessage);
            
            switch (message.type) {
                case 'chat_message':
                    this.handleChatMessage(clientId, message);
                    break;
                case 'join_room':
                    this.handleJoinRoom(clientId, message);
                    break;
                case 'leave_room':
                    this.handleLeaveRoom(clientId, message);
                    break;
                case 'broadcast':
                    this.handleBroadcast(clientId, message);
                    break;
                default:
                    console.log(`Unknown message type: ${message.type}`);
            }
        } catch (error) {
            console.error('Error handling message:', error);
        }
    }
    
    handleChatMessage(clientId, message) {
        const room = message.room;
        const roomClients = this.rooms.get(room);
        
        if (roomClients) {
            const messageData = {
                type: 'chat_message',
                clientId,
                message: message.content,
                timestamp: new Date().toISOString()
            };
            
            roomClients.forEach(client => {
                if (client.ws.readyState === WebSocket.OPEN) {
                    client.ws.send(JSON.stringify(messageData));
                }
            });
        }
    }
    
    handleJoinRoom(clientId, message) {
        const room = message.room;
        
        if (!this.rooms.has(room)) {
            this.rooms.set(room, new Map());
        }
        
        const client = this.clients.get(clientId);
        const roomClients = this.rooms.get(room);
        
        client.rooms.add(room);
        roomClients.set(clientId, client);
        
        this.send(clientId, { type: 'joined_room', room });
    }
    
    handleLeaveRoom(clientId, message) {
        const room = message.room;
        
        if (this.rooms.has(room)) {
            this.rooms.get(room).delete(clientId);
        }
        
        const client = this.clients.get(clientId);
        client.rooms.delete(room);
    }
    
    handleBroadcast(clientId, message) {
        const client = this.clients.get(clientId);
        
        this.clients.forEach((c, id) => {
            if (id !== clientId && c.ws.readyState === WebSocket.OPEN) {
                c.ws.send(JSON.stringify({
                    type: 'broadcast',
                    from: clientId,
                    message: message.content
                }));
            }
        });
    }
    
    handleDisconnect(clientId) {
        const client = this.clients.get(clientId);
        
        client.rooms.forEach(room => {
            if (this.rooms.has(room)) {
                this.rooms.get(room).delete(clientId);
            }
        });
        
        this.clients.delete(clientId);
        console.log(`Client disconnected: ${clientId}`);
    }
    
    send(clientId, message) {
        const client = this.clients.get(clientId);
        if (client && client.ws.readyState === WebSocket.OPEN) {
            client.ws.send(JSON.stringify(message));
        }
    }
    
    generateClientId() {
        return `client_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
    
    start() {
        this.wss.listen(this.port, () => {
            console.log(`WebSocket server started on port ${this.port}`);
        });
    }
}

module.exports = WebSocketServer;

class WebSocketClient {
    constructor(url, options = {}) {
        this.url = url;
        this.reconnectInterval = options.reconnectInterval || 1000;
        this.maxReconnectAttempts = options.maxReconnectAttempts || 5;
        this.reconnectAttempts = 0;
        
        this.handlers = {
            open: [],
            close: [],
            error: [],
            message: []
        };
        
        this.connect();
    }
    
    connect() {
        this.ws = new WebSocket(this.url);
        
        this.ws.onopen = (event) => {
            console.log('WebSocket connected');
            this.reconnectAttempts = 0;
            this.handlers.open.forEach(handler => handler(event));
        };
        
        this.ws.onclose = (event) => {
            console.log('WebSocket closed');
            this.handlers.close.forEach(handler => handler(event));
            this.attemptReconnect();
        };
        
        this.ws.onerror = (event) => {
            console.error('WebSocket error:', event);
            this.handlers.error.forEach(handler => handler(event));
        };
        
        this.ws.onmessage = (event) => {
            try {
                const data = JSON.parse(event.data);
                this.handlers.message.forEach(handler => handler(data));
            } catch (e) {
                this.handlers.message.forEach(handler => handler(event.data));
            }
        };
    }
    
    attemptReconnect() {
        if (this.reconnectAttempts < this.maxReconnectAttempts) {
            this.reconnectAttempts++;
            console.log(`Attempting reconnect (${this.reconnectAttempts}/${this.maxReconnectAttempts})`);
            setTimeout(() => this.connect(), this.reconnectInterval);
        } else {
            console.error('Max reconnection attempts reached');
        }
    }
    
    send(data) {
        if (this.ws && this.ws.readyState === WebSocket.OPEN) {
            const message = typeof data === 'string' ? data : JSON.stringify(data);
            this.ws.send(message);
        } else {
            console.error('WebSocket is not connected');
        }
    }
    
    on(event, handler) {
        if (this.handlers[event]) {
            this.handlers[event].push(handler);
        }
    }
    
    off(event, handler) {
        if (this.handlers[event]) {
            const index = this.handlers[event].indexOf(handler);
            if (index > -1) {
                this.handlers[event].splice(index, 1);
            }
        }
    }
    
    close() {
        if (this.ws) {
            this.ws.close();
        }
    }
    
    readyState() {
        return this.ws ? this.ws.readyState : WebSocket.CLOSED;
    }
}

class WebSocketManager {
    constructor() {
        this.clients = new Map();
        this.messageHandlers = new Map();
    }
    
    createClient(id, url, options) {
        const client = new WebSocketClient(url, options);
        
        client.on('message', (data) => {
            const handler = this.messageHandlers.get(id);
            if (handler) {
                handler(data);
            }
        });
        
        this.clients.set(id, client);
        return client;
    }
    
    sendToClient(id, data) {
        const client = this.clients.get(id);
        if (client) {
            client.send(data);
        }
    }
    
    broadcast(data) {
        this.clients.forEach(client => {
            client.send(data);
        });
    }
    
    removeClient(id) {
        const client = this.clients.get(id);
        if (client) {
            client.close();
            this.clients.delete(id);
        }
    }
    
    onMessage(id, handler) {
        this.messageHandlers.set(id, handler);
    }
}

import { useState, useEffect, useRef, useCallback } from 'react';

const useWebSocket = (url, options = {}) => {
    const [isConnected, setIsConnected] = useState(false);
    const [lastMessage, setLastMessage] = useState(null);
    const [error, setError] = useState(null);
    
    const wsRef = useRef(null);
    const optionsRef = useRef(options);
    
    const connect = useCallback(() => {
        if (wsRef.current?.readyState === WebSocket.OPEN) {
            return;
        }
        
        const ws = new WebSocket(url);
        
        ws.onopen = () => {
            setIsConnected(true);
            setError(null);
            optionsRef.current.onOpen?.();
        };
        
        ws.onclose = (event) => {
            setIsConnected(false);
            optionsRef.current.onClose?.(event);
            
            if (!event.wasClean && optionsRef.current.reconnect) {
                setTimeout(connect, optionsRef.current.reconnectInterval || 3000);
            }
        };
        
        ws.onerror = (event) => {
            setError(event);
            optionsRef.current.onError?.(event);
        };
        
        ws.onmessage = (event) => {
            const message = optionsRef.current.parseMessage?.(event.data) || event.data;
            setLastMessage(message);
            optionsRef.current.onMessage?.(message);
        };
        
        wsRef.current = ws;
    }, [url]);
    
    useEffect(() => {
        connect();
        
        return () => {
            if (wsRef.current) {
                wsRef.current.close();
            }
        };
    }, [connect]);
    
    const sendMessage = useCallback((message) => {
        if (wsRef.current?.readyState === WebSocket.OPEN) {
            const data = optionsRef.current.stringifyMessage?.(message) || message;
            wsRef.current.send(data);
        }
    }, []);
    
    const disconnect = useCallback(() => {
        if (wsRef.current) {
            wsRef.current.close();
        }
    }, []);
    
    return {
        isConnected,
        lastMessage,
        error,
        sendMessage,
        disconnect,
        reconnect: connect
    };
};

export default useWebSocket;

class WebSocketSubprotocol:
    CHAT = "chat.v1"
    NOTIFICATIONS = "notifications.v1"
    Realtime_API = "realtime.v1"
    MQTT = "mqtt"

class WebSocketSubprotocolHandler:
    def __init__(self, protocol: str):
        self.protocol = protocol
        self.handlers = {
            WebSocketSubprotocol.CHAT: self.handle_chat,
            WebSocketSubprotocol.NOTIFICATIONS: self.handle_notification,
            WebSocketSubprotocol.Realtime_API: self.handle_realtime,
        }
    
    async def handle_message(self, websocket, message: dict):
        handler = self.handlers.get(self.protocol)
        if handler:
            await handler(websocket, message)
    
    async def handle_chat(self, websocket, message: dict):
        # Chat-specific handling
        pass
    
    async def handle_notification(self, websocket, message: dict):
        # Notification-specific handling
        pass
    
    async def handle_realtime(self, websocket, message: dict):
        # Real-time API handling
        pass

class WebSocketExtensions:
    PERMESSAGE_DEFLATE = "permessage-deflate"
    MULTIPLEXING = "websocket-extensions"

class CompressionHandler:
    def __init__(self):
        self.compressor = None
    
    def enable_compression(self):
        return {
            'sec-websocket-extensions': 'permessage-deflate; ' \
                                       'client_max_window_bits; ' \
                                       'server_max_window_bits=15'
        }
    
    def compress_message(self, data: bytes) -> bytes:
        # Compression logic
        return data
    
    def decompress_message(self, data: bytes) -> bytes:
        # Decompression logic
        return data

import secrets
import hashlib
import hmac
from typing import Optional

class WebSocketAuthenticator:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
    
    def generate_token(self, user_id: str, expires_in: int = 3600) -> str:
        expiry = int(time.time()) + expires_in
        payload = f"{user_id}:{expiry}"
        
        signature = hmac.new(
            self.secret_key.encode(),
            payload.encode(),
            hashlib.sha256
        ).hexdigest()
        
        return f"{payload}:{signature}"
    
    def verify_token(self, token: str) -> Optional[str]:
        try:
            user_id, expiry, signature = token.rsplit(':', 2)
            
            expected_signature = hmac.new(
                self.secret_key.encode(),
                f"{user_id}:{expiry}".encode(),
                hashlib.sha256
            ).hexdigest()
            
            if not hmac.compare_digest(signature, expected_signature):
                return None
            
            if int(expiry) < int(time.time()):
                return None
            
            return user_id
        
        except ValueError:
            return None

class WebSocketAuthMiddleware:
    def __init__(self, app, auth: WebSocketAuthenticator):
        self.app = app
        self.auth = auth
    
    async def __call__(self, websocket, path):
        token = websocket.request_headers.get('Sec-WebSocket-Protocol')
        
        if not token:
            await websocket.close(4001, "Authentication required")
            return
        
        user_id = self.auth.verify_token(token)
        
        if not user_id:
            await websocket.close(4002, "Invalid token")
            return
        
        websocket.user_id = user_id
        
        await self.app(websocket, path)

class OriginValidator:
    def __init__(self, allowed_origins: list):
        self.allowed_origins = allowed_origins
    
    async def validate(self, origin: str) -> bool:
        if not origin:
            return False
        
        if "*" in self.allowed_origins:
            return True
        
        return origin in self.allowed_origins
    
    async def handle_request(self, websocket, path):
        origin = websocket.request_headers.get('Origin')
        
        if not await self.validate(origin):
            await websocket.close(4003, "Origin not allowed")
            return False
        
        return True

import ssl

class WebSocketSSLConfig:
    @staticmethod
    def create_ssl_context(cert_file: str, key_file: str) -> ssl.SSLContext:
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(cert_file, key_file)
        context.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20')
        
        return context
    
    @staticmethod
    def create_wss_config(cert_file: str, key_file: str) -> dict:
        ssl_context = WebSocketSSLConfig.create_ssl_context(cert_file, key_file)
        
        return {
            'ssl': ssl_context
        }

import asyncio
from typing import Dict, List

class WebSocketConnectionPool:
    def __init__(self, max_connections: int = 10000):
        self.max_connections = max_connections
        self.connections: Dict[str, WebSocketConnection] = {}
        self.rooms: Dict[str, set] = {}
        self._lock = asyncio.Lock()
    
    async def add_connection(self, connection_id: str, websocket):
        async with self._lock:
            if len(self.connections) >= self.max_connections:
                raise Exception("Connection pool full")
            
            self.connections[connection_id] = WebSocketConnection(
                id=connection_id,
                websocket=websocket
            )
    
    async def remove_connection(self, connection_id: str):
        async with self._lock:
            if connection_id in self.connections:
                connection = self.connections[connection_id]
                
                for room in list(connection.rooms):
                    await self.leave_room(connection_id, room)
                
                del self.connections[connection_id]
    
    async def join_room(self, connection_id: str, room: str):
        async with self._lock:
            if connection_id not in self.connections:
                return
            
            if room not in self.rooms:
                self.rooms[room] = set()
            
            self.rooms[room].add(connection_id)
            self.connections[connection_id].rooms.add(room)
    
    async def leave_room(self, connection_id: str, room: str):
        async with self._lock:
            if room in self.rooms:
                self.rooms[room].discard(connection_id)
            
            if connection_id in self.connections:
                self.connections[connection_id].rooms.discard(room)
    
    async def broadcast_to_room(self, room: str, message: bytes):
        if room not in self.rooms:
            return
        
        await asyncio.gather(
            *[
                self.connections[conn_id].send(message)
                for conn_id in self.rooms[room]
                if conn_id in self.connections
            ],
            return_exceptions=True
        )
    
    async def get_connection_count(self) -> int:
        async with self._lock:
            return len(self.connections)
    
    async def get_room_count(self, room: str) -> int:
        async with self._lock:
            return len(self.rooms.get(room, set()))

@dataclass
class WebSocketConnection:
    id: str
    websocket: any
    rooms: set = field(default_factory=set)
    metadata: dict = field(default_factory=dict)

import redis.asyncio as redis
import json
from typing import Optional

class RedisWebSocketManager:
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)
        self.pubsub = self.redis.pubsub()
    
    async def register_connection(self, connection_id: str, server_id: str):
        await self.redis.sadd(f"server:{server_id}:connections", connection_id)
        await self.redis.hset(
            f"connection:{connection_id}",
            mapping={
                "server_id": server_id,
                "connected_at": str(time.time())
            }
        )
    
    async def unregister_connection(self, connection_id: str):
        server_id = await self.redis.hget(f"connection:{connection_id}", "server_id")
        
        if server_id:
            await self.redis.srem(f"server:{server_id.decode()}:connections", connection_id)
        
        await self.redis.delete(f"connection:{connection_id}")
    
    async def publish_message(self, channel: str, message: dict):
        await self.redis.publish(channel, json.dumps(message))
    
    async def subscribe_to_channel(self, channel: str):
        await self.pubsub.subscribe(channel)
    
    async def get_connection_server(self, connection_id: str) -> Optional[str]:
        server_id = await self.redis.hget(f"connection:{connection_id}", "server_id")
        return server_id.decode() if server_id else None
    
    async def get_all_server_ids(self) -> List[str]:
        servers = await self.redis.keys("server:*:connections")
        return [s.decode().split(":")[1] for s in servers]

BEST_PRACTICES = {
    "connection_lifecycle": [
        "Implement heartbeat/ping-pong to detect stale connections",
        "Set appropriate connection timeouts",
        "Handle reconnection gracefully on client side",
        "Clean up resources on disconnect",
        "Limit concurrent connections per user/IP"
    ],
    
    "message_handling": [
        "Always validate and sanitize incoming messages",
        "Use message type dispatching for organization",
        "Implement message acknowledgment for critical operations",
        "Handle message queuing during brief disconnections",
        "Use binary protocol for large messages"
    ],
    
    "security": [
        "Always use WSS for production",
        "Implement authentication during handshake",
        "Validate Origin header",
        "Implement rate limiting per connection",
        "Sanitize all user input"
    ],
    
    "performance": [
        "Use connection pooling where applicable",
        "Implement message batching for high throughput",
        "Compress messages when possible",
        "Monitor connection state and health",
        "Use efficient serialization (Protocol Buffers, MessagePack)"
    ],
    
    "scalability": [
        "Design for horizontal scaling from the start",
        "Use Redis pub/sub for cross-server communication",
        "Implement sticky sessions or message routing",
        "Monitor and alert on connection metrics",
        "Plan for graceful degradation under load"
    ]
}

class WebSocketErrorHandler:
    @staticmethod
    async def handle_connection_error(websocket, error):
        logger.error(f"Connection error: {error}")
        
        try:
            await websocket.close(1011, "Internal server error")
        except:
            pass
    
    @staticmethod
    async def handle_message_error(websocket, message_id, error):
        logger.error(f"Message error ({message_id}): {error}")
        
        await websocket.send(json.dumps({
            'type': 'error',
            'message_id': message_id,
            'error': str(error)
        }))
    
    @staticmethod
    async def handle_timeout_error(websocket, operation):
        logger.warning(f"Timeout during {operation}")
        
        await websocket.send(json.dumps({
            'type': 'timeout',
            'operation': operation
        }))

import pytest
import asyncio
import websockets

@pytest.mark.asyncio
async def test_websocket_connection():
    async with websockets.connect("ws://localhost:8765") as websocket:
        await websocket.send('{"type": "ping"}')
        
        response = await asyncio.wait_for(websocket.recv(), timeout=5)
        data = json.loads(response)
        
        assert data['type'] == 'pong'

@pytest.mark.asyncio
async def test_websocket_chat():
    async with websockets.connect("ws://localhost:8765") as ws1:
        async with websockets.connect("ws://localhost:8765") as ws2:
            await ws1.send(json.dumps({
                'type': 'join_room',
                'room': 'test-room'
            }))
            
            response = await ws1.recv()
            assert json.loads(response)['type'] == 'room_joined'
            
            await ws1.send(json.dumps({
                'type': 'chat_message',
                'room': 'test-room',
                'message': 'Hello'
            }))
            
            response = await ws2.recv()
            data = json.loads(response)
            
            assert data['type'] == 'chat_message'
            assert data['message'] == 'Hello'

@pytest.mark.asyncio
async def test_websocket_reconnection():
    client = WebSocketClient("ws://localhost:8765", {
        'reconnect': True,
        'reconnectInterval': 1000,
        'maxReconnectAttempts': 3
    })
    
    await asyncio.sleep(2)
    
    assert client.readyState() == WebSocket.OPEN

class NotificationServer:
    def __init__(self, server: WebSocketServer):
        self.server = server
        self._register_handlers()
    
    def _register_handlers(self):
        self.server.register_handler('subscribe', self.handle_subscribe)
        self.server.register_handler('unsubscribe', self.handle_unsubscribe)
    
    async def handle_subscribe(self, websocket, data: Dict):
        user_id = data.get('user_id')
        categories = data.get('categories', ['all'])
        
        for category in categories:
            room = f"notifications:{category}"
            await self.server.join_room(websocket, room)
        
        await websocket.send(json.dumps({
            'type': 'subscribed',
            'categories': categories
        }))
    
    async def send_notification(self, user_id: str, notification: Dict):
        await self.server.send_to_room(f"notifications:{user_id}", {
            'type': 'notification',
            'notification': notification,
            'timestamp': datetime.utcnow().isoformat()
        })
    
    async def broadcast_notification(self, category: str, notification: Dict):
        await self.server.send_to_room(f"notifications:{category}", {
            'type': 'notification',
            'notification': notification,
            'timestamp': datetime.utcnow().isoformat()
        })

class CollaborativeEditor:
    def __init__(self, server: WebSocketServer):
        self.server = server
        self.documents: Dict[str, str] = {}
        self.cursors: Dict[str, Dict] = {}
        self._register_handlers()
    
    def _register_handlers(self):
        self.server.register_handler('open_document', self.handle_open)
        self.server.register_handler('edit_document', self.handle_edit)
        self.server.register_handler('cursor_move', self.handle_cursor)
        self.server.register_handler('close_document', self.handle_close)
    
    async def handle_open(self, websocket, data: Dict):
        doc_id = data.get('document_id')
        
        await self.server.join_room(websocket, f"doc:{doc_id}")
        
        content = self.documents.get(doc_id, "")
        
        await websocket.send(json.dumps({
            'type': 'document_content',
            'document_id': doc_id,
            'content': content,
            'cursors': self._get_cursors(doc_id)
        }))
    
    async def handle_edit(self, websocket, data: Dict):
        doc_id = data.get('document_id')
        edit = data.get('edit')
        
        self._apply_edit(doc_id, edit)
        
        await self.server.send_to_room(f"doc:{doc_id}", {
            'type': 'document_edit',
            'document_id': doc_id,
            'edit': edit,
            'editor': websocket.remote_address
        })
    
    async def handle_cursor(self, websocket, data: Dict):
        doc_id = data.get('document_id')
        position = data.get('position')
        
        self.cursors[f"{doc_id}:{websocket.remote_address}"] = position
    
    def _apply_edit(self, doc_id: str, edit: Dict):
        if doc_id not in self.documents:
            self.documents[doc_id] = ""
        
        operation = edit.get('operation')
        
        if operation == 'insert':
            pos = edit.get('position')
            text = edit.get('text')
            self.documents[doc_id] = (
                self.documents[doc_id][:pos] + 
                text + 
                self.documents[doc_id][pos:]
            )
    
    def _get_cursors(self, doc_id: str) -> Dict:
        return {
            k.split(':')[1]: v 
            for k, v in self.cursors.items() 
            if k.startswith(f"{doc_id}:")
        }