WebSocket Consumers

WebSocket Consumers

WebSocketConsumer turns a WebSocket endpoint into a class with three lifecycle hooks. It runs the receive loop for you, decodes each message according to a declared encoding, and gives every connection a Channel so it can join groups and broadcast.

a chat consumer

from sillo.websockets import WebSocketConsumer

class ChatConsumer(WebSocketConsumer):
    encoding = "json"

    async def on_connect(self, websocket):
        await websocket.accept()
        await self.join_group("lobby")

    async def on_receive(self, websocket, data):
        await self.broadcast({"said": data}, group_name="lobby")

    async def on_disconnect(self, websocket, close_code):
        await self.leave_group("lobby")


app.add_ws_route(ChatConsumer.as_route("/ws/chat"))

That is a working multi-user chat room in fifteen lines. The rest of this page is about what happens when it is not a demo.

When to reach for a consumer

Use a consumer when the endpoint has real structure: several message types, group membership, per-connection state, or cleanup that has to run on disconnect. The class gives that a shape, and the receive loop is written once rather than in every handler.

Use a plain ws_route function for endpoints that do one thing — an echo service, a log tail, a single-purpose notification stream. A class with one overridden method is not an improvement over a function.

The lifecycle

__call__ drives the whole connection:

1. Builds a Channel wrapping the socket
2. Calls on_connect(websocket)
3. Loops on websocket.receive(), decoding and dispatching to on_receive(websocket, data)
4. Breaks on websocket.disconnect, recording the close code
5. Calls on_disconnect(websocket, close_code) in a finally

on_connect must accept the connection

The default implementation calls await websocket.accept(). Override it and you own that responsibility:

the override that breaks everything

async def on_connect(self, websocket):
    await self.join_group("lobby")      # never accepted — nothing works

Every subsequent receive_* raises RuntimeError: WebSocket is not connected. Need to call "accept" first. Call accept() first, or call await super().on_connect(websocket).

Conversely, calling accept() twice — once in super() and once in your override — raises on the second call, because the state machine has already left CONNECTING.

on_disconnect always runs

It is in a finally, so it runs after a clean disconnect, after an exception in on_receive, and after an exception in on_connect. That makes it the right place to leave groups and release resources.

It also means on_disconnect can run when on_connect never finished. If on_connect raised before joining a group, the matching leave_group runs anyway — which, as it happens, is a path into a crash. See the next section.

Exceptions are re-raised, not swallowed

sillo/websockets/consumers.py

try:
    ...                     # the receive loop
except Exception as exc:
    close_code = status.WS_1011_INTERNAL_ERROR
    raise exc
finally:
    await self.on_disconnect(self.websocket, close_code)

An unhandled exception in on_receive sets the close code to 1011, runs on_disconnect, and then propagates. close_code is assigned but the socket is not explicitly closed here — the surrounding error middleware handles that. The practical consequence is that one malformed message can terminate a connection. Catch what you can handle:

keeping the connection alive through bad input

async def on_receive(self, websocket, data):
    try:
        await self.dispatch(data)
    except ValidationError as exc:
        await websocket.send_json({"error": str(exc)})
    except Exception:
        self.logger.exception("handler failed")
        await websocket.send_json({"error": "internal error"})

Encoding

Set encoding and decode() parses each message before on_receive sees it.

encoding	data is	On mismatch
"json"	the parsed object	closes with 1003, raises RuntimeError("Malformed JSON data received.")
"text"	str	closes with 1003 if a binary frame arrives
"bytes"	bytes	closes with 1003 if a text frame arrives
None (default)	text if present, else bytes	KeyError if the frame has neither

The JSON branch accepts either frame type — it decodes a binary frame as UTF-8 before parsing — so a client that sends JSON over binary frames works without configuration.

Malformed JSON kills the connection

With encoding = "json", a single unparseable message closes the socket with 1003 and raises. There is no hook to handle it, because the failure happens in decode() before on_receive is called.

If clients might send garbage — anything driven by a browser extension, a flaky mobile client, or a hand-written reconnect loop — leave encoding unset and parse in your handler, where you can reply with an error instead of hanging up:

class Tolerant(WebSocketConsumer):
    async def on_receive(self, websocket, data):
        try:
            payload = json.loads(data)
        except json.JSONDecodeError:
            await websocket.send_json({"error": "invalid JSON"})
            return
        await self.dispatch(payload)

Channels and groups

Every consumer instance gets self.channel — a Channel wrapping the socket, with a JSON payload type when encoding == "json" and text otherwise.

the group helpers

await self.join_group("room:42")
await self.leave_group("room:42")
await self.broadcast(payload, group_name="room:42", save_history=True)
await self.send_to(channel_uuid, payload)
channels = await self.group("room:42")     # list[Channel]

broadcast sends to every channel in the group including the sender. If you want everyone else, filter:

broadcasting to everyone but the sender

async def on_receive(self, websocket, data):
    for channel in await self.group("room:42"):
        if channel.uuid != self.channel.uuid:
            await channel._send({"said": data})

That reaches into a private method, which is a sign the public API is missing an exclude parameter. It works, and it is the only option today.

leave_group can raise UnboundLocalError

ChannelBox.remove_channel_from_group assigns its return status inside two conditional branches. When the channel is not in the named group and the group still has other members, neither branch runs and the function raises:

UnboundLocalError: cannot access local variable 'channel_remove_status'
where it is not associated with a value

This is reachable from ordinary code. A consumer whose on_connect failed before joining, or one that leaves a group twice, will hit it — and because leave_group is normally called from on_disconnect, the error surfaces during cleanup where it is least expected.

Guard the call until this is fixed upstream:

async def on_disconnect(self, websocket, close_code):
    try:
        await self.leave_group("lobby")
    except UnboundLocalError:
        pass

Or track membership yourself and only leave groups you know you joined.

The 3600-second channel expiry

__call__ constructs the channel with expires=3600 — hard-coded, not configurable through the class.

self.channel = Channel(websocket=self.websocket, expires=3600, payload_type=...)

ChannelBox._clean_expired() runs on every remove_channel_from_group call and drops channels whose expiry has passed. Channel._send resets created, so the clock is an idle timer rather than an absolute one: a channel that sends or receives a broadcast at least once an hour never expires.

The failure mode is a quiet connection. A presence indicator or a notification socket that receives nothing for an hour gets silently removed from its groups while the underlying socket stays open. The client believes it is subscribed; the server has forgotten it.

Two mitigations. Send a heartbeat from the server at an interval well under an hour — which also keeps intermediary proxies from closing idle connections. Or construct the channel yourself in on_connect with expires=None and manage group membership directly, bypassing the consumer’s channel entirely.

Registering a consumer

as_route

app.add_ws_route(ChatConsumer.as_route("/ws/chat"))
app.add_ws_route(RoomConsumer.as_route("/ws/room/{room_id}"))

as_route returns a WebsocketRoute whose handler instantiates the class per connection. That per-connection instance is why self.channel and any instance attributes you set are safe — they are not shared between clients. Class attributes are shared; do not accumulate state on the class.

Path parameters come off the socket, not the constructor:

path parameters in a consumer

class RoomConsumer(WebSocketConsumer):
    encoding = "json"

    async def on_connect(self, websocket):
        await websocket.accept()
        self.room = f"room:{websocket.path_params['room_id']}"
        await self.join_group(self.room)

    async def on_receive(self, websocket, data):
        await self.broadcast(data, group_name=self.room)

    async def on_disconnect(self, websocket, close_code):
        await self.leave_group(self.room)

Storing the group name on self in on_connect and reading it in on_disconnect is the pattern worth copying — it keeps the join and the leave in sync when the name is computed.

The middleware class attribute exists on WebSocketConsumer and is never read by any code path. Do not populate it expecting anything to happen; apply cross-cutting concerns in on_connect or in ASGI middleware.

Logging

Consumers log at INFO by default: connection established, every message received, every broadcast, every group join and leave.

consumer = ChatConsumer()          # logging_enabled=True, logger=sillo logger

Because on_receive’s default logs the full message body, a chat application at default settings writes every private message to your logs in plaintext. Turn it off for anything carrying user content:

quieting a consumer

class ChatConsumer(WebSocketConsumer):
    def __init__(self):
        super().__init__(logging_enabled=False)

Note that as_route calls cls() with no arguments, so the constructor override above is how you configure it — there is no way to pass arguments through as_route.

A complete example

A room-scoped chat with authentication, structured messages, presence, and cleanup that does not crash.

a production-shaped consumer

import json

from sillo.websockets import WebSocketConsumer, status


class ChatConsumer(WebSocketConsumer):
    def __init__(self):
        super().__init__(logging_enabled=False)
        self.room = None
        self.user = None

    async def on_connect(self, websocket):
        self.user = await authenticate(websocket.query_params.get("ticket"))
        if self.user is None:
            await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
            return

        await websocket.accept()
        self.room = f"room:{websocket.path_params['room_id']}"
        await self.join_group(self.room)
        await self.broadcast(
            {"type": "presence", "event": "join", "user": self.user.name},
            group_name=self.room,
        )

    async def on_receive(self, websocket, data):
        try:
            payload = json.loads(data)
        except (json.JSONDecodeError, TypeError):
            await websocket.send_json({"type": "error", "detail": "invalid JSON"})
            return

        if payload.get("type") != "message":
            await websocket.send_json({"type": "error", "detail": "unknown type"})
            return

        text = str(payload.get("text", ""))[:2000]
        await self.broadcast(
            {"type": "message", "user": self.user.name, "text": text},
            group_name=self.room,
            save_history=True,
        )

    async def on_disconnect(self, websocket, close_code):
        if self.room is None:
            return
        try:
            await self.leave_group(self.room)
        except UnboundLocalError:
            pass
        await self.broadcast(
            {"type": "presence", "event": "leave", "user": self.user.name},
            group_name=self.room,
        )

Note the self.room is None guard: on_disconnect runs even when on_connect rejected the connection, and without the guard the cleanup path would fail on a rejected client.

What not to do

Do not override on_connect without accepting. Every receive will raise.

Do not call accept() twice. The state machine raises.

Do not use encoding = "json" with untrusted clients. One bad message closes the socket.

Do not call leave_group unguarded. It can raise UnboundLocalError.

Do not assume broadcast excludes the sender. It does not.

Do not leave default logging on for user content. Every message body is logged at INFO.

Do not populate the middleware attribute. Nothing reads it.

Do not keep per-connection state on the class. Instances are per-connection; class attributes are shared.

Do not let a quiet connection idle past an hour without a heartbeat, or it silently leaves its groups.

Testing a consumer

Consumers are testable through TestClient exactly like function handlers, and because the class holds its state on the instance, two concurrent test connections do not interfere.

testing broadcast between two clients

def test_broadcast_reaches_both_clients():
    with TestClient(app) as client:
        with client.websocket_connect("/ws/room/1?ticket=t1") as a:
            with client.websocket_connect("/ws/room/1?ticket=t2") as b:
                a.receive_json()          # a's own presence event
                b.receive_json()          # b sees its join
                a.send_json({"type": "message", "text": "hi"})
                assert b.receive_json()["text"] == "hi"

Asserting on the sender’s stream is how you catch the broadcast-includes-sender behaviour early — if you expect a not to receive its own message, that test will tell you otherwise.

Performance notes

One consumer instance and one asyncio task per connection. The instance is cheap; what it holds is not — a loaded user object per connection, at ten thousand connections, is ten thousand user objects.

broadcast iterates the group and awaits each send sequentially. A slow client delays every client behind it in the iteration. For large groups, dispatch the sends concurrently with asyncio.gather over a snapshot of the channel list, and put a timeout on each.

send_to scans every group and every channel looking for a UUID match — O(total connections) per call. For direct messaging, keep your own {user_id: channel} dictionary instead.

API reference

Member	Signature	Notes
encoding	"json" / "text" / "bytes" / None	Class attribute
on_connect	(websocket)	Must accept; default does
on_receive	(websocket, data)	data is already decoded
on_disconnect	(websocket, close_code)	Always runs
join_group / leave_group	(group_name)	leave_group may raise UnboundLocalError
broadcast	(payload, group_name="default", save_history=False)	Includes the sender
send_to	(channel_id: UUID, payload)	Linear scan of all groups
group	(group_name) -> list[Channel]
as_route	(path) -> WebsocketRoute	Instantiates with no arguments
self.channel	Channel	Hard-coded expires=3600

Related

• WebSockets — the raw connection and routing
• Channels — what self.channel actually is
• Groups — ChannelBox and broadcast semantics
• Events — dispatching by message type
• Middleware — where cross-cutting concerns actually go