Socket Controllers — Aquilia Documentation
Comprehensive guide and documentation for Socket Controllers in the Aquilia framework. View API reference, examples, and implementation patterns.
WebSockets / Socket Controllers Socket Controllers WebSocket handlers in Aquilia are declared inside classes inheriting from SocketController . Every incoming client connection gets a stateless controller instance bound to its own request-scoped DI container, letting you inject services, handle connection events, validate schemas, and stream responses. Extended Implementation Example The following controller demonstrates connection handshakes, pub/sub room subscriptions, validation schema boundaries, rate-limiting guards, and message acknowledgments in a single production-ready class: from aquilia.sockets import ( SocketController, Socket, OnConnect, OnDisconnect, Event, AckEvent, Subscribe, Unsubscribe, Guard, Connection, Schema ) from aquilia.di import Inject from typing import Annotated @Socket( path="/rooms/:room_id", allowed_origins=["https://myapp.com"], max_message_size=1024 * 1024, # 1MB compression=True ) class ChatRoomController(SocketController): def __init__(self, chat_service: Annotated[ChatService, Inject()]): self.chat = chat_service @OnConnect async def on_connect(self, conn: Connection): # Path parameter extraction room_id = conn.scope.path_params.get("room_id") # 1. Join connection to room await conn.join(room_id) # 2. Retrieve identity properties user = conn.identity username = user.username if user else "Anonymous" await conn.send_event("presence.join", ) @Subscribe("chat.history") async def on_subscribe(self, conn: Connection): room_id = conn.scope.path_params["room_id"] history = await self.chat.get_history(room_id) await conn.send_event("chat.history_dump", ) @Event("message.send", schema=Schema( )) async def on_new_message(self, conn: Connection, payload: dict): room_id = conn.scope.path_params["room_id"] # Save through dependency-injected service msg = await self.chat.save_message(room_id, conn.id, payload["text"]) # Broadcast to all connections in the room await conn.broadcast(room_id, "message.receive", msg.to_dict()) @AckEvent("typing.state") async def on_typing(self, conn: Connection, payload: dict): room_id = conn.scope.path_params["room_id"] # Broadcast to room excluding the current sender await conn.broadcast( room_id, "typing.state", , exclude_connection=conn.id ) # Return ACK status back to the sender client return @Unsubscribe("chat.history") async def on_unsubscribe(self, conn: Connection): pass @OnDisconnect async def on_disconnect(self, conn: Connection, reason: str | None = None): room_id = conn.scope.path_params.get("room_id") await conn.leave(room_id) await conn.broadcast(room_id, "presence.leave", ) Decorator Reference , )` }, closed connection. " f"Reason: . " f"Duration: s" ) # 3. Clean up cluster states await conn.leave("broadcast_lobby")` }, ), ack=False ) async def on_new_comment(self, conn: Connection, payload: dict): post_id = payload["post_id"] comment_text = payload["comment"] # Save payload to database... await self.save_comment(conn.identity.id, post_id, comment_text) # Broadcast updates to the room await conn.broadcast(f"post_ ", "comment.added", )` }, ) ) async def on_checkout(self, conn: Connection, payload: dict) -> dict: cart_id = payload["cart_id"] # Process order workflow... success, transaction_id = await self.order_service.checkout(cart_id) if not success: # Returned dict is automatically packed into ACK payload back to sender return return ` }, ) return # Join billing events channel room await conn.join("room:billing_alerts") await conn.send_event("subscribe.success", )` }, )` }, )` } ].map((item, i) => ( ))} Connection Object API Every controller method receives a Connection instance representing the active client session: )` }, , Closing: ") # Set and read arbitrary thread-safe connection local attributes state.custom_attrs["last_ping_at"] = datetime.utcnow() await conn.send_event("pong.state", )` }, , ) return await conn.send_event("profile.data", )` }, ) return # Read/Write directly to the ASGI Session transport counter = session.get("counter", 0) + 1 session["counter"] = counter await conn.send_event("session.count", )` }, )` }, , ack=True ) # Log msg_id to map incoming user acknowledgments later logger.info(f"Dispatched message with tracking ID: ")` }, } envelope # Sends a raw JSON array or dictionary directly down the socket await conn.send_json([ , ])` }, , " # Join connection to the channel await conn.join(room_name) # Broadcast join event to all other members in the room await conn.broadcast( room=room_name, event="user.joined", payload= , exclude_connection=conn.id )` }, " # Leave room channel await conn.leave(room_name) # Broadcast leave notification to remaining members await conn.broadcast( room=room_name, event="user.left", payload= )` }, ) # Triggers WS 1008 Policy Violation close event await conn.disconnect(reason="forced_termination", code=1008)` }, ].map((item, i) => ( ))} WebSocket Chunked Streaming AquilaSockets supports first-class streaming. If a handler returns an AsyncIterator or Iterator, the runtime will automatically consume it and stream chunks to the client: @Event("logs.stream") async def handle_logs_stream(self, conn: Connection, payload: dict) -> AsyncIterator[dict]: # Generator yielding chunks back to the client for i in range(10): await asyncio.sleep(0.5) yield "} # Handled by runtime: # Sends 'logs.stream.chunk' events carrying each dictionary. # Ends the stream by dispatching 'logs.stream.end'. Overview WebSocket Runtime )
Go to Homepage