WebSocket Gateway — Complete Guide
In this tutorial, you'll learn about Websocket Gateway. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
WebSocket gateways handle the upgrade from HTTP to WebSocket protocol, manage long-lived connections, route messages to appropriate backends, and scale across multiple instances.
What You'll Learn
By the end of this lesson, you will implement WebSocket connection upgrade, message routing, pub-sub patterns, and understand scaling strategies for WebSocket gateways.
Why It Matters
WebSockets require persistent connections that traditional HTTP gateways may not support. A WebSocket-aware gateway properly handles upgrades and maintains connection state.
Real-World Use
A chat application uses the gateway to upgrade HTTP to WebSocket, route messages to the chat service, and broadcast to other connected clients.
WebSocket Gateway Flow
sequenceDiagram
Client->>Gateway: HTTP Upgrade Request
Gateway->>Gateway: Validate & Upgrade
Gateway->>Backend: Forward WebSocket
Client->>Gateway: Message
Gateway->>Backend: Route Message
Backend->>Gateway: Response
Gateway-->>Client: Forward Response
WebSocket Upgrade Handler
# websocket_upgrade.py
from typing import Dict, Optional, Set
import hashlib
import base64
class WebSocketUpgrader:
MAGIC_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
def can_upgrade(self, headers: Dict[str, str]) -> bool:
upgrade = headers.get("Upgrade", "").lower()
connection = headers.get("Connection", "").lower()
ws_key = headers.get("Sec-WebSocket-Key")
return (
upgrade == "websocket"
and "upgrade" in connection
and ws_key is not None
)
def perform_upgrade(self, headers: Dict[str, str]) -> Optional[str]:
if not self.can_upgrade(headers):
return None
ws_key = headers.get("Sec-WebSocket-Key", "")
accept_key = base64.b64encode(
hashlib.sha1((ws_key + self.MAGIC_GUID).encode()).digest()
).decode()
return accept_key
def route_to_backend(self, path: str, backends: Dict[str, str]) -> Optional[str]:
for prefix, backend in backends.items():
if path.startswith(prefix):
return backend
return None
upgrader = WebSocketUpgrader()
backends = {
"/ws/chat": "chat-service:8080",
"/ws/notifications": "notif-service:8080",
"/ws/game": "game-service:8080",
}
requests = [
("/ws/chat", {"Upgrade": "websocket", "Connection": "Upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ=="}),
("/ws/game", {"Upgrade": "websocket", "Connection": "Upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ=="}),
("/api/data", {"Upgrade": "websocket", "Connection": "Upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ=="}),
]
for path, headers in requests:
accept = upgrader.perform_upgrade(headers)
backend = upgrader.route_to_backend(path, backends)
if accept and backend:
print(f"{path:25s} -> Upgrade OK, route to {backend}")
elif accept:
print(f"{path:25s} -> Upgrade OK, no backend route")
else:
print(f"{path:25s} -> No upgrade (not a WebSocket request)")
Expected output:
/ws/chat -> Upgrade OK, route to chat-service:8080
/ws/game -> Upgrade OK, route to game-service:8080
/api/data -> No upgrade (not a WebSocket request)
WebSocket Message Router
# ws_message_router.py
import json
import time
from typing import Any, Callable, Dict, List, Optional
class WebSocketMessageRouter:
def __init__(self):
self.handlers: Dict[str, List[Callable]] = {}
self.connections: Dict[str, dict] = {}
def on(self, event: str, handler: Callable):
if event not in self.handlers:
self.handlers[event] = []
self.handlers[event].append(handler)
def register_connection(self, conn_id: str, user_id: str, metadata: Dict = None):
self.connections[conn_id] = {
"user_id": user_id,
"metadata": metadata or {},
"connected_at": time.time(),
"subscribed_channels": set(),
}
def subscribe(self, conn_id: str, channel: str):
if conn_id in self.connections:
self.connections[conn_id]["subscribed_channels"].add(channel)
def unsubscribe(self, conn_id: str, channel: str):
if conn_id in self.connections:
self.connections[conn_id]["subscribed_channels"].discard(channel)
def route_message(self, conn_id: str, message: str) -> List[str]:
try:
data = json.loads(message)
event = data.get("event", "message")
except json.JSONDecodeError:
event = "message"
responses = []
for handler in self.handlers.get(event, []):
result = handler(conn_id, data if 'data' in locals() else message)
if result:
responses.append(result)
return responses
def disconnect(self, conn_id: str):
self.connections.pop(conn_id, None)
router = WebSocketMessageRouter()
@router.on("chat:send")
def handle_chat(conn_id, data):
return {"event": "chat:broadcast", "data": data, "from": conn_id}
@router.on("game:move")
def handle_game(conn_id, data):
return {"event": "game:update", "data": data, "player": conn_id}
router.register_connection("conn_1", "user_1")
router.register_connection("conn_2", "user_2")
responses = router.route_message("conn_1", '{"event": "chat:send", "text": "Hello!"}')
for response in responses:
print(f"Response: {response}")
responses = router.route_message("conn_2", '{"event": "game:move", "position": [1, 2]}')
for response in responses:
print(f"Response: {response}")
Expected output:
Response: {'event': 'chat:broadcast', 'data': {'event': 'chat:send', 'text': 'Hello!'}, 'from': 'conn_1'}
Response: {'event': 'game:update', 'data': {'event': 'game:move', 'position': [1, 2]}, 'player': 'conn_2'}
Common Mistakes
1. No Heartbeat Mechanism
Without heartbeats, dead connections accumulate. Implement ping/pong to detect and close stale connections.
2. Shared State on Single Instance
WebSocket connections are tied to the instance that accepted them. Use a pub-sub backend (Redis) to broadcast across instances.
3. Unlimited Connections Per Client
A malicious client can open unlimited connections. Set per-client connection limits at the gateway.
4. Not Handling Graceful Disconnect
Abrupt disconnects leave resources allocated. Handle close frames and cleanup connection state.
5. No Authentication for Upgrade
Allow unauthenticated WebSocket upgrades. Authenticate during the HTTP upgrade request before switching protocols.
Practice Questions
1. How does the gateway upgrade HTTP to WebSocket?
The client sends an Upgrade header. The gateway validates the Sec-WebSocket-Key, computes the Sec-WebSocket-Accept response, and switches protocols.
2. Why are WebSocket connections harder to scale?
WebSockets are stateful and long-lived. A client connection is tied to one gateway instance, requiring a pub-sub layer for cross-instance messaging.
3. What is the purpose of WebSocket heartbeats?
Heartbeats detect dead connections so the gateway can clean up resources and remove stale entries.
4. How does the gateway route WebSocket messages?
After upgrade, the gateway forwards raw WebSocket frames to the appropriate backend service based on the original URL path.
Challenge
Design a WebSocket gateway architecture that handles 100K concurrent connections, supports pub-sub messaging across instances via Redis, and includes authentication, Rate Limiting, and heartbeat management.
FAQ
Mini Project: WebSocket Gateway Simulator
# ws_gateway.py
import hashlib
import base64
import json
import time
from typing import Dict, List, Optional
class WebSocketGateway:
def __init__(self):
self.connections: Dict[str, dict] = {}
self.backend_routes: Dict[str, str] = {}
self.max_connections_per_client = 5
def add_route(self, path: str, backend: str):
self.backend_routes[path] = backend
def upgrade(self, conn_id: str, path: str, headers: Dict) -> dict:
upgrade = headers.get("Upgrade", "").lower()
if upgrade != "websocket":
return {"status": 400, "error": "Not a WebSocket upgrade"}
backend = self._find_backend(path)
if not backend:
return {"status": 404, "error": "No backend for path"}
client_ip = headers.get("X-Forwarded-For", "unknown")
client_connections = sum(
1 for c in self.connections.values() if c.get("client_ip") == client_ip
)
if client_connections >= self.max_connections_per_client:
return {"status": 429, "error": "Too many connections"}
self.connections[conn_id] = {
"path": path,
"backend": backend,
"client_ip": client_ip,
"connected_at": time.time(),
"messages_sent": 0,
"messages_received": 0,
}
return {"status": 101, "backend": backend, "conn_id": conn_id}
def _find_backend(self, path: str) -> Optional[str]:
for prefix, backend in sorted(self.backend_routes.items(), key=lambda x: -len(x[0])):
if path.startswith(prefix):
return backend
return None
def receive_message(self, conn_id: str, message: str) -> Optional[str]:
conn = self.connections.get(conn_id)
if not conn:
return None
conn["messages_received"] += 1
try:
data = json.loads(message)
data["_backend"] = conn["backend"]
data["_conn_id"] = conn_id
return json.dumps(data)
except json.JSONDecodeError:
return message
def disconnect(self, conn_id: str):
return self.connections.pop(conn_id, None)
gw = WebSocketGateway()
gw.add_route("/ws/chat", "chat-backend:8080")
gw.add_route("/ws/game", "game-backend:8080")
result = gw.upgrade("conn_1", "/ws/chat",
{"Upgrade": "websocket", "X-Forwarded-For": "10.0.0.1"})
print(f"Upgrade result: {result['status']} -> {result.get('backend', 'N/A')}")
routed = gw.receive_message("conn_1", '{"text": "Hello"}')
print(f"Routed message: {routed}")
gw.disconnect("conn_1")
print(f"After disconnect, connections: {len(gw.connections)}")
Expected output:
Upgrade result: 101 -> chat-backend:8080
Routed message: {"text": "Hello", "_backend": "chat-backend:8080", "_conn_id": "conn_1"}
After disconnect, connections: 0
What's Next
You understand WebSocket gateway patterns. Next, learn about Kong API gateway, then explore NGINX as API gateway.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro