WebSocketConnectionManager delegates storage to a ConnectionBackend
implementation. Swap in your own to support clustering or observability.
from falcon_pachinko.websocket import ConnectionBackend, WebSocketConnectionManager
class RedisBackend(ConnectionBackend):
...
conn_mgr = WebSocketConnectionManager(backend=RedisBackend(...))
app.ws_connection_manager = conn_mgr # before router mounts
Backend requirements (see ConnectionBackend ABC):
- Raise
ValueErroron duplicateadd_connection. - Make
leave_roomandremove_connectionidempotent. - Provide read-only
websockets/roomsviews and a consistentsnapshot. - Ignore or document stale room memberships in
snapshot.
Tests illustrating this pattern live in:
tests/test_connection_manager_unit.py::test_manager_uses_custom_backendtests/behaviour/custom_connection_backend.feature