Adapters — Aquilia Documentation
Comprehensive guide and documentation for Adapters in the Aquilia framework. View API reference, examples, and implementation patterns.
WebSockets / Adapters Scaling Adapters Adapters route events, channel subscriptions, room memberships, and messages across your application processes. This pub/sub interface enables AquilaSockets to scale horizontally from a single node to multi-server clusters. The Adapter Base Class To write a custom adapter (e.g. for NATS or RabbitMQ), inherit from the base Adapter class and implement the following async signature interfaces: from aquilia.sockets import Adapter, MessageEnvelope class CustomAdapter(Adapter): async def initialize(self) -> None: """Establish client connections to external message brokers.""" pass async def shutdown(self) -> None: """Close connections and flush buffers during server teardown.""" pass async def publish(self, namespace: str, room: str, envelope: MessageEnvelope, exclude_connection: str | None = None) -> None: """Publish a message to all members subscribing to a room across clusters.""" pass async def broadcast(self, namespace: str, envelope: MessageEnvelope, exclude_connection: str | None = None) -> None: """Broadcast a message to all active connections inside a namespace.""" pass async def join_room(self, namespace: str, room: str, connection_id: str) -> None: """Register a connection ID as a member of a room.""" pass async def leave_room(self, namespace: str, room: str, connection_id: str) -> None: """Unregister a connection ID from a room.""" pass async def get_room_members(self, namespace: str, room: str) -> set[str]: """Fetch active connection IDs inside a room across nodes.""" return set() async def get_connection_count(self, namespace: str) -> int: """Fetch total client count within the namespace.""" return 0 InMemoryAdapter (Default) The default adapter manages connection states, room memberships, and broadcasts entirely inside the local application memory. It is optimized for single-instance applications, developer environments, and automated testing suites. from aquilia.sockets import AquilaSockets, InMemoryAdapter sockets = AquilaSockets( router=router, adapter=InMemoryAdapter() ) RedisAdapter (Production Scaling) The RedisAdapter uses Redis Pub/Sub channels to sync message delivery between server processes. Connection metadata and room rosters are maintained atomically inside Redis Sets and Sorted Sets: from aquilia.sockets import AquilaSockets, RedisAdapter adapter = RedisAdapter( url="redis://localhost:6379/0", channel_prefix="ws:", pool_size=15 ) sockets = AquilaSockets( router=router, adapter=adapter ) Implementing a Custom Adapter To integrate with other messaging brokers like NATS or RabbitMQ, inherit from Adapter: from aquilia.sockets import Adapter, MessageEnvelope class NatsWebSocketAdapter(Adapter): def __init__(self, nats_url: str): self.nats_url = nats_url self.nc = None async def initialize(self): import nats self.nc = await nats.connect(self.nats_url) async def publish(self, namespace: str, room: str, envelope: MessageEnvelope, exclude_connection: str | None = None): subject = f"ws. . " payload = self.codec.encode(envelope) await self.nc.publish(subject, payload) async def shutdown(self): if self.nc: await self.nc.close() WebSocket Runtime Templates )
Go to Homepage