Broadcasting¶

Introduction¶

In many modern web applications, WebSockets are used to implement real-time, live-updating interfaces. When some data is updated on the server, a message is sent over a WebSocket connection to be handled by the client. Arvel's broadcasting publishes named events on named channels through a pluggable driver, so your frontend can subscribe and react in real time.

Configuration¶

Broadcasting reads config/broadcasting.py when present; the BROADCASTING_* environment variables are the fallback for any key the file doesn't set (see the cascade):

BROADCASTING_DEFAULT=redis-pubsub

Drivers¶

Registering the Provider¶

Broadcasting is opt-in. Add BroadcastServiceProvider to bootstrap/providers.py. It binds the Broadcast facade; without it, the facade raises a runtime error.

Broadcasting Events¶

The ShouldBroadcast Contract¶

To make an event broadcastable, implement the ShouldBroadcast contract. It defines what channels to broadcast on, the event name, and the payload:

from collections.abc import Mapping, Sequence
from arvel.events.event import Event
from arvel.broadcasting.should_broadcast import ShouldBroadcast


class OrderShipped(Event, ShouldBroadcast):
    order_id: int

    def broadcast_on(self) -> Sequence[str]:
        return [f"orders.{self.order_id}"]

    def broadcast_as(self) -> str:
        return "order.shipped"

    def broadcast_with(self) -> Mapping[str, object]:
        return {"order_id": self.order_id}

Broadcasting From an Event¶

When you dispatch an event that implements ShouldBroadcast, the dispatcher pushes it to the broadcast driver automatically — after the synchronous listeners finish:

from arvel.facades.event import Event

await Event.dispatch(OrderShipped(order_id=1))   # listeners run, then it broadcasts

Broadcasting Directly¶

You can also publish without an event, straight through the facade:

from arvel.facades.broadcast import Broadcast

await Broadcast.send(
    channels=["orders.1"],
    event="order.shipped",
    payload={"order_id": 1},
)

# Or push a ShouldBroadcast object explicitly:
await Broadcast.event(OrderShipped(order_id=1))

Channels & Authorization¶

Register an authorization callback for a channel pattern with the Broadcast.channel decorator. The callback decides whether a given user may listen on a channel:

@Broadcast.channel("orders.{order_id}")
async def authorize_order(user: User, order_id: int) -> bool:
    return await Order.where(id=order_id, user_id=user.id).exists()

Testing¶

BroadcasterFake records every broadcast(...) call so you can assert what was published, with no real driver involved. It exposes calls and assert_broadcasted(event_name):

from arvel.testing.broadcasting import BroadcasterFake

fake = BroadcasterFake()
await fake.broadcast(["orders.1"], "order.shipped", {"order_id": 1})

fake.assert_broadcasted("order.shipped")
assert fake.calls[0].payload == {"order_id": 1}