Topics and streaming

Topic

Topic(
    name: str,
    maxsize: int = DEFAULT_MAXSIZE,
    policy: str = DROP_OLDEST,
    backend: Any = None,
)

A named fan-out point shared by every worker loop in the process.

With a backend it is also shared across processes: emitting appends to a Redis stream, and a tail task in every other process feeds its local subscribers. The publishing process delivers locally itself and the tail skips its own node, so nobody sees a message twice.

Source code in python/aether/_streams.py

def __init__(
    self,
    name: str,
    maxsize: int = DEFAULT_MAXSIZE,
    policy: str = DROP_OLDEST,
    backend: Any = None,
) -> None:
    if policy not in POLICIES:
        raise ValueError(
            f"unknown policy {policy!r}; choose one of {', '.join(sorted(POLICIES))}"
        )
    if maxsize < 1:
        raise ValueError("maxsize must be at least 1")
    self.name = name
    self.maxsize = maxsize
    self.policy = policy
    self.backend = backend
    self._subs: list[Subscription] = []
    self._lock = threading.Lock()
    self._tail: Any = None

name instance-attribute

name = name

maxsize instance-attribute

maxsize = maxsize

policy instance-attribute

policy = policy

backend instance-attribute

backend = backend

subscribers property

subscribers: int

Local subscribers only. Other processes are not visible from here.

durable property

durable: bool

history async

history(count: int = 100) -> list

Recent messages, oldest first. Durable topics only.

Source code in python/aether/_streams.py

async def history(self, count: int = 100) -> list:
    """Recent messages, oldest first. Durable topics only."""
    if self.backend is None:
        raise RuntimeError(f"topic {self.name!r} is not durable, so it has no history")
    return await self.backend.history(self.name, count=count)

consumer

consumer(group: str, name: str, **options: Any)

A member of a consumer group, for at-least-once processing.

Unlike subscribe, which is a broadcast to everyone, each message goes to exactly one member of the group and stays pending until acked.

Source code in python/aether/_streams.py

def consumer(self, group: str, name: str, **options: Any):
    """A member of a consumer group, for at-least-once processing.

    Unlike `subscribe`, which is a broadcast to everyone, each message goes
    to exactly one member of the group and stays pending until acked.
    """
    if self.backend is None:
        raise RuntimeError(
            f"topic {self.name!r} is not durable; consumer groups need a backend"
        )
    from ._redis import Consumer

    return Consumer(self.backend, self.name, group, name, **options)

subscribe

subscribe(
    maxsize: int | None = None, policy: str | None = None
) -> Subscription

Start receiving. Must be called from inside a running event loop.

The subscription binds to the loop it was created on, which is how a producer on another worker knows where to deliver.

Source code in python/aether/_streams.py

def subscribe(self, maxsize: int | None = None, policy: str | None = None) -> Subscription:
    """Start receiving. Must be called from inside a running event loop.

    The subscription binds to the loop it was created on, which is how a
    producer on another worker knows where to deliver.
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        raise RuntimeError(
            "subscribe() needs a running event loop; call it inside a handler"
        ) from None

    policy = policy or self.policy
    if policy not in POLICIES:
        raise ValueError(
            f"unknown policy {policy!r}; choose one of {', '.join(sorted(POLICIES))}"
        )
    sub = Subscription(self, maxsize or self.maxsize, policy, loop)
    with self._lock:
        self._subs.append(sub)
    self._ensure_tail()
    return sub

emit async

emit(item: Any) -> int

Deliver to every local subscriber. Returns how many received it.

On a durable topic the message is appended to the stream first, so that returning means it is recorded and other processes will see it. That costs a round trip, which is the trade being made by asking for durability.

Only awaits on a local subscriber when it uses the block policy and is full.

Source code in python/aether/_streams.py

async def emit(self, item: Any) -> int:
    """Deliver to every local subscriber. Returns how many received it.

    On a durable topic the message is appended to the stream *first*, so
    that returning means it is recorded and other processes will see it.
    That costs a round trip, which is the trade being made by asking for
    durability.

    Only awaits on a local subscriber when it uses the `block` policy and
    is full.
    """
    if self.backend is not None:
        await self.backend.publish(self.name, item)

    delivered = 0
    for sub in self._snapshot():
        if not sub._try_offer(item):
            await sub._offer(item)
        delivered += 1
    return delivered

emit_nowait

emit_nowait(item: Any) -> int

Deliver to local subscribers without ever waiting.

A block subscriber that is full is treated as drop_newest, because the alternative here would be blocking a thread that must not block.

Refused on a durable topic: appending to the stream is an await, so this could only ever deliver locally, and a call named emit that silently skipped durability is worse than an error.

Source code in python/aether/_streams.py

def emit_nowait(self, item: Any) -> int:
    """Deliver to local subscribers without ever waiting.

    A `block` subscriber that is full is treated as `drop_newest`, because
    the alternative here would be blocking a thread that must not block.

    Refused on a durable topic: appending to the stream is an await, so this
    could only ever deliver locally, and a call named `emit` that silently
    skipped durability is worse than an error.
    """
    if self.backend is not None:
        raise RuntimeError(
            f"topic {self.name!r} is durable; use `await emit()` so the "
            f"message is recorded, not just delivered in this process"
        )
    delivered = 0
    for sub in self._snapshot():
        if not sub._try_offer(item):
            sub.dropped += 1
        delivered += 1
    return delivered

close

close() -> None

Source code in python/aether/_streams.py

def close(self) -> None:
    with self._lock:
        tail, self._tail = self._tail, None
    if tail is not None:
        tail.cancel()
    for sub in self._snapshot():
        sub.close()

Subscription

Subscription(topic: Topic, maxsize: int, policy: str, loop)

One subscriber's view of a topic. Async-iterable, and closeable.

Source code in python/aether/_streams.py

def __init__(self, topic: "Topic", maxsize: int, policy: str, loop) -> None:
    self.topic = topic
    self.maxsize = maxsize
    self.policy = policy
    self.dropped = 0
    self._buffer: deque[Any] = deque()
    self._loop = loop
    # Producers may run on other worker loops, so buffer and waiter state
    # are guarded rather than relying on any atomicity of deque itself.
    self._lock = threading.Lock()
    self._getter: asyncio.Future | None = None
    self._putters: deque[tuple[Any, asyncio.Future]] = deque()
    self._closed = False

topic instance-attribute

topic = topic

maxsize instance-attribute

maxsize = maxsize

policy instance-attribute

policy = policy

dropped instance-attribute

dropped = 0

closed property

closed: bool

pending property

pending: int

close

close() -> None

Stop the iterator and release anyone waiting on it.

Source code in python/aether/_streams.py

def close(self) -> None:
    """Stop the iterator and release anyone waiting on it."""
    with self._lock:
        if self._closed:
            return
        self._closed = True
        getter, self._getter = self._getter, None
        putters = list(self._putters)
        self._putters.clear()

    if getter is not None:
        self._loop.call_soon_threadsafe(_resolve, getter)
    for loop, waiter in putters:
        loop.call_soon_threadsafe(_resolve, waiter)
    self.topic._remove(self)

TopicFull

Bases: Exception

A subscriber's buffer is full and its policy is error.

Backpressure policies

DROP_OLDEST module-attribute

DROP_OLDEST = 'drop_oldest'

DROP_NEWEST module-attribute

DROP_NEWEST = 'drop_newest'

BLOCK module-attribute

BLOCK = 'block'

ERROR module-attribute

ERROR = 'error'

Server-Sent Events

SSE

SSE(
    source: Any,
    *,
    ping: float | None = 15.0,
    status: int = 200,
)

A streaming text/event-stream response.

ping sends a comment line when idle that long, which stops proxies and load balancers from closing an idle connection. None disables it.

Source code in python/aether/_sse.py

def __init__(self, source: Any, *, ping: float | None = 15.0, status: int = 200) -> None:
    """`ping` sends a comment line when idle that long, which stops proxies
    and load balancers from closing an idle connection. None disables it."""
    if not hasattr(source, "__aiter__"):
        raise TypeError(
            f"SSE needs an async iterable, got {type(source).__name__}. "
            f"A topic subscription or an async generator both work"
        )
    self.source = source
    self.ping = ping
    self.status = status

source instance-attribute

source = source

ping instance-attribute

ping = ping

status instance-attribute

status = status

Event dataclass

Event(
    data: Any,
    event: str | None = None,
    id: str | None = None,
    retry: int | None = None,
)

One event, when the defaults are not enough.

Yield plain values for the common case; yield this to set a name, an id for resumption, or a client retry hint.

event and id may not contain a line break or a NUL; both raise ValueError. Checked here so the traceback points at the code that built the event, and again at render time, because this is a mutable dataclass and the fields can be reassigned afterwards.

data instance-attribute

data: Any

event class-attribute instance-attribute

event: str | None = None

id class-attribute instance-attribute

id: str | None = None

retry class-attribute instance-attribute

retry: int | None = None

WebSocket

WebSocket

WebSocket(core: Any)

An open connection. Async-iterable over incoming messages.

Source code in python/aether/_websocket.py

def __init__(self, core: Any) -> None:
    self._core = core
    self._loop = asyncio.get_running_loop()

closed property

closed: bool

receive async

receive() -> str | bytes | None

Next message, or None once the peer has closed.

Source code in python/aether/_websocket.py

async def receive(self) -> str | bytes | None:
    """Next message, or None once the peer has closed."""
    while True:
        message = self._core.try_receive()
        if message is not None:
            return message
        if self._core.closed:
            return None

        waiter = self._loop.create_future()
        # `notify` fires immediately if something arrived in the meantime,
        # so this cannot miss a message that landed during the check above.
        self._core.notify(self._loop, lambda: _resolve(waiter))
        await waiter

receive_json async

receive_json() -> Any

Source code in python/aether/_websocket.py

async def receive_json(self) -> Any:
    message = await self.receive()
    if message is None:
        raise WebSocketClosed("connection closed while waiting for a message")
    if isinstance(message, bytes):
        message = message.decode()
    return json.loads(message)

send async

send(data: Any) -> None

Send a message.

str goes as text and bytes as binary. Anything else is serialized to JSON, which covers dicts and pydantic models.

Source code in python/aether/_websocket.py

async def send(self, data: Any) -> None:
    """Send a message.

    `str` goes as text and `bytes` as binary. Anything else is serialized
    to JSON, which covers dicts and pydantic models.
    """
    if isinstance(data, str):
        sent = self._core.send_text(data)
    elif isinstance(data, (bytes, bytearray, memoryview)):
        sent = self._core.send_bytes(bytes(data))
    elif is_model_instance(data):
        # Text, not binary: a model is JSON, and a dict sent the same way
        # would arrive as text. Frame type should not depend on which.
        sent = self._core.send_text(to_json(data).decode())
    else:
        sent = self._core.send_text(json.dumps(data, separators=(",", ":")))

    if not sent:
        raise WebSocketClosed("socket is closed or its send buffer is full")

send_json async

send_json(data: Any) -> None

Source code in python/aether/_websocket.py

async def send_json(self, data: Any) -> None:
    await self.send(data)

close async

close() -> None

Start a clean close handshake.

Source code in python/aether/_websocket.py

async def close(self) -> None:
    """Start a clean close handshake."""
    self._core.close()

WebSocketClosed

Bases: Exception

A send was attempted after the peer went away.