sillo-wire

Rooms, presence and fan-out for Sillo WebSockets.

pip install sillo-wire

Installs as sillo-wire, imports as sillo.wire.

from sillo import SilloApp
from sillo.wire import Hub, Peer

app = SilloApp()
hub = Hub()

@app.ws_route("/ws/room/{name}")
async def room(socket, name: str):
    await socket.accept()
    peer = Peer(socket, identity=socket.query_params.get("user"))
    await hub.join(peer, name)
    try:
        async for message in socket.iter_json():
            await hub.broadcast(name, message)
    finally:
        await hub.disconnect(peer)

Why this exists

Three things differ from the obvious implementation, and they are the whole point of the package.

A broadcast never blocks. Writing straight to each socket in turn means the slowest member of a room sets the pace for everyone else — a client that has stopped reading fills its kernel buffer, the write blocks, and the rest of the room waits behind it. Here every peer has a bounded queue and a writer task, so a broadcast only ever enqueues:

report = await hub.broadcast("lobby", {"msg": "hello"})
report.delivered   # 41
report.dropped     #  2   queues were full
report.failed      #  1   socket was already gone

You get a DeliveryReport rather than nothing, because a fan-out you cannot measure is a fan-out you cannot operate.

Nothing is global. A Hub is an ordinary object. Two of them are two independent worlds, so tests get a fresh one per case instead of remembering to flush shared state, and a multi-tenant application keeps traffic apart without a naming convention.

History is replayable. Every envelope carries a monotonic sequence, so a client that reconnects asks for what it missed rather than for everything or for nothing:

await hub.replay(peer, "lobby", since=last_seq_the_client_saw)

Slow consumers

When a peer's queue fills, what happens is a choice, not a default:

from sillo.wire import Overflow, Peer

Peer(socket, overflow=Overflow.DROP_OLDEST)   # keep current — prices, cursors
Peer(socket, overflow=Overflow.DROP_NEWEST)   # keep order — reconcile later
Peer(socket, overflow=Overflow.CLOSE)         # disconnect and let it reconnect

Presence

@hub.on_join
async def joined(room, peer):
    await hub.broadcast(room, {"event": "joined", "who": peer.identity})

hub.identities("lobby")   # ["ada", "bob"] — people, not sockets
hub.count("lobby")        # 5 — subscriptions

Two peers can share an identity — the same person with a phone and two tabs — and send_to reaches all of them:

await hub.send_to("ada", {"notice": "your export is ready"})

Consumers

RoomConsumer is the class-based form. It accepts the socket, builds the peer, joins the rooms, pumps messages, and guarantees the peer is removed from every room when the connection ends — including when a hook raises.

from sillo.wire import Hub, RoomConsumer

hub = Hub()

class Chat(RoomConsumer):
    hub = hub

    async def identify(self, ctx):
        return ctx.query_params.get("user")

    async def rooms(self, ctx):
        return [ctx.path_params["room"]]

    async def on_message(self, data):
        await self.broadcast({"from": self.peer.identity, "text": data})

app.add_ws_route(path="/ws/{room}", handler=Chat.as_handler())

Backlog

Retention is per room and capped by payload bytes, evicting oldest first:

from sillo.wire import Hub, MemoryBacklog, NullBacklog

Hub(backlog=MemoryBacklog(capacity_bytes=4 * 1024 * 1024))
Hub(backlog=NullBacklog())    # keep nothing — typing indicators, telemetry

Backlog is a Protocol, so a Redis or Postgres store satisfies it without importing anything from here.

Testing

sillo.wire.testing ships the piece unit tests are missing — a socket:

from sillo.wire import Hub, Peer
from sillo_wire.testing import FakeSocket, drain

async def test_a_broadcast_reaches_the_room():
    hub, socket = Hub(), FakeSocket()
    peer = Peer(socket)
    await hub.join(peer, "lobby")

    await hub.broadcast("lobby", {"hello": True})
    await drain(peer)          # broadcasts enqueue; this waits for the write

    assert socket.sent == [{"hello": True}]

FakeSocket(delay=…) simulates a client that is slow to read, and FakeSocket(fail=True) one that has gone away — the two cases that are hardest to reproduce against a real server and the two most worth testing.

Reference

The two import paths

sillo.wire and sillo_wire name the same objects. The code lives in the top-level sillo_wire package; sillo.wire is an alias, so it reads as part of the framework:

from sillo.wire import Hub     # both of these
from sillo_wire import Hub     # bind the same class

The alias is a meta-path finder registered by a .pth at interpreter startup — the only hook that runs before an import sillo.wire could fail. Type checkers never run import hooks, so they are served separately by the partial stubs in sillo-stubs/ (PEP 561), which are additive: mypy resolves sillo.wire and still uses the framework's own inline types for the rest of sillo.

Nothing is written into the framework's package directory. Shipping sillo/wire/ in there would be simpler, and it is what this did first — but two distributions sharing one directory goes wrong in both directions. Installing the framework from a checkout moves where sillo resolves and orphans the copy in site-packages; removing or replacing the framework leaves that directory standing with no __init__.py, which is an override rather than an addition. Uninstalling either package here leaves the other exactly as it was.

Working on it

The alias works under an editable install too — the .pth is shipped by the editable build target as well as the wheel.

pip install -e ".[dev]"
pytest --cov            # 100% required, bootstrap included
ruff check sillo_wire tests _sillo_wire_bootstrap.py
mypy sillo_wire

Requirements

Python 3.10+, sillo-framework 0.3 or newer. No other dependencies.

Licence

BSD-3-Clause.