Toolbox

In-process pub/sub with Relay. It does not replace a message broker or production observability.

Relay

Relay is in-process only: one Python process, dotted subject strings, and wildcard patterns on subscribe. It is not a message broker. There is no persistence, replay, or cross-process delivery. For multiple workers or hosts, use NATS, Redis, or another broker.

Subjects and patterns

publish(subject, data) requires an exact dotted subject (non-empty segments, no wildcards, no leading or trailing dots).

subscribe(*patterns) accepts:

an exact dotted subject (chat.room1),
a trailing prefix wildcard (chat.*),
or the catch-all *.

subscribe() validates patterns and merges overlaps so each publish delivers at most once per subscription. Invalid patterns raise StarioError.

If no subscriber is registered for any matching pattern, publish is a no-op for fan-out. Messages published before async with relay.subscribe(...) as sub: are never delivered — enter the subscription before any publish the handler must receive.

prefix.* matches subjects with a dot after the prefix. A publish to exactly chat matches exact chat and *, not chat.*.

Subscription lifecycle

relay.subscribe(...) returns a RelaySubscription handle. You must enter it with async with before receiving messages. The handle buffers nothing until __aenter__ runs.

Consume with await sub.receive() or async for msg in sub on the same entered handle. Consume only on the asyncio loop that entered the subscription; concurrent receive() calls raise StarioRuntime. For long-lived async for loops, combine with c.alive() or check c.closing during shutdown (Runtime — Shutdown).

Bare async for subject, payload in relay.subscribe("chat.*") does not work — the iterable requires an active subscription and raises StarioRuntime if you skip async with.

When ordering matters, enter async with relay.subscribe(...) before any publish you need to receive. publish is synchronous and thread-safe; it does not await subscribers.

python

import asyncio

from stario import Relay

relay = Relay[dict]()

async def main():

    async with relay.subscribe("chat.*") as sub:

        relay.publish("chat.control", {"kind": "joined"})

        subject, payload = await sub.receive()

        assert subject == "chat.control"

        relay.publish("chat.room1", {"text": "hi"})

        async for subject, payload in sub:

            if subject == "chat.room1":

                break

Each subscription remembers the asyncio loop that registered it. publish from another thread schedules delivery on that loop with call_soon_threadsafe. The inbox is unbounded; a slow consumer can grow memory without limit.

Typical uses: fan-out between handlers and SSE loops in one process, cache-invalidation hints, tests on the same event loop.

class Relay()

In-process publish/subscribe registry.

Use async with relay.subscribe(...) as sub:, then consume with sub.receive() or async for msg in sub. Nothing is queued before enter.

Dead subscribers (closed event loop) are removed silently; publish does not report delivery failures.

Relay.drop_subscription(subscription)

Relay.publish(subject, data)

Deliver (subject, data) to every subscriber whose pattern matches.

Thread-safe from any OS thread. Delivery runs after the subscriber snapshot is taken; neither Relay.lock nor _inbox_lock is held while Future.set_result runs. Does not await subscribers.