WebSockets — Aquilia Documentation
Comprehensive guide and documentation for WebSockets in the Aquilia framework. View API reference, examples, and implementation patterns.
Advanced / WebSockets WebSockets Overview AquilaSockets provides production-grade WebSocket support featuring a declarative, decorator-driven syntax. Every connection is backed by its own request-scoped DI container, auth-first upgrade guards, structured message envelopes, event streaming, and horizontal scaling via message broker adapters. System Integration & Registration To activate WebSocket routing, a controller must be mounted in your module's manifest, which is in turn loaded by the workspace configuration. 1. Workspace Registration Register your module within the workspace builder in workspace.py: from aquilia.workspace import Workspace, Module workspace = ( Workspace("myapp") .runtime(port=8000) .module( Module("chat") .route_prefix("/chat") ) ) 2. Manifest Mounting Mount the socket controller class inside your module's manifest file in the socket_controllers parameter: from aquilia.manifest import AppManifest manifest = AppManifest( name="chat", version="1.0.0", controllers=[], socket_controllers=[ "modules.chat.controllers:RoomChatController", ], services=[ "modules.chat.services:ChatService" ] ) Subsystem Architecture The WebSocket module separates connection lifecycle, message codec parsing, security validation, and pub/sub distribution into distinct components: , , , , , ].map((item, i) => ( ))} Extended Controller Implementation WebSocket controllers are defined by inheriting from SocketController and decorating with @Socket : from aquilia.sockets import ( SocketController, Socket, OnConnect, OnDisconnect, Event, AckEvent, Subscribe, Connection, Schema ) from aquilia import Inject @Socket("/ws/chat/:room") class RoomChatController(SocketController): @Inject() def __init__(self, chat_service: ChatService): self.chat = chat_service @OnConnect async def handle_connect(self, conn: Connection): # Read parameters from ASGI path patterns room = conn.scope.path_params.get("room") await conn.join(room) # Inject user identity resolved from handshake auth user = conn.identity await conn.send_event("welcome", !" }) @Event("chat.message", schema=Schema( )) async def on_message(self, conn: Connection, payload: dict): room = conn.scope.path_params["room"] # Broadcast to all connections in the room await conn.broadcast(room, "chat.message", ) @AckEvent("user.typing") async def on_typing(self, conn: Connection, payload: dict): room = conn.scope.path_params["room"] await conn.broadcast(room, "user.typing", ) return # Returned directly to the sender as an ACK Core Capabilities DI-Scoped Handlers Every connection creates its own request-scoped dependency injection container. Scoped objects are automatically created, injected, and cleaned up when the client disconnects. Robust Handshake Security Authenticates connections via Authorization Bearer headers, query string tokens, or cookies, using HTTP guard logic before upgrading the connection to a WebSocket. Horizontal Scaling Pluggable broker adapters (such as RedisAdapter ) forward broadcast and room events across multiple servers seamlessly, ensuring instant pub/sub delivery. Cache Socket Controllers )
Go to Homepage