pamoja.bus

Idiomatic event-bus facade.

One publisher, many subscribers, inside a single process. It is how the parts of a gateway talk to each other without knowing about each other, so a sampler can announce a reading and whatever cares about readings picks it up.

A subscriber only sees events published after it existed, so subscribe before publishing anything it needs to see. An EventBus endpoint publishes and receives; an EventPublisher only publishes, so a part that announces and never reads holds one of those.

 1"""Idiomatic event-bus facade.
 2
 3One publisher, many subscribers, inside a single process. It is how the parts of
 4a gateway talk to each other without knowing about each other, so a sampler can
 5announce a reading and whatever cares about readings picks it up.
 6
 7A subscriber only sees events published after it existed, so subscribe before
 8publishing anything it needs to see. An ``EventBus`` endpoint publishes and
 9receives; an ``EventPublisher`` only publishes, so a part that announces and never
10reads holds one of those.
11"""
12
13from __future__ import annotations
14
15from pamoja._native import EventBus, EventPublisher
16
17__all__ = ["EventBus", "EventPublisher"]
class EventBus:

One endpoint on an event bus.

An endpoint both publishes and receives, and it receives what it publishes itself. Each subscriber needs its own, taken with subscribe, because an endpoint only sees events published after it existed. Only next_event and next_text wait: publishing and subscribing return at once, even while a wait on the same endpoint is open.

def subscribe(self, /):

Takes another endpoint on the same bus.

The new endpoint sees events published from now on, not those already sent, so subscribe before publishing anything it needs to see.

def publisher(self, /):

Takes a publish-only handle on the same bus, for a part that announces and never reads.

def publish(self, /, event):

Publishes an event to every subscriber, this endpoint included: bytes, or text such as an event name.

It never waits: a subscriber that has fallen behind loses its oldest event rather than holding up the publisher.

def next_event(self, /):

Waits for the next event on this endpoint.

The wait lasts until an event arrives. asyncio.wait_for gives up sooner, and the wait it cancels takes no event.

def next_text(self, /):

Waits for the next event as text.

Raises ValueError if the event is not UTF-8 text.

missed

How many events this endpoint lost by falling behind, as of its last completed wait.

class EventPublisher:

A publish-only handle to an event bus.

It has no queue of its own, so a part that only announces never fills a buffer it does not read. Publishing never waits, so a callback on another thread can call it without an event loop.

def publish(self, /, event):

Hands an event to every current subscriber: bytes, or text such as an event name.

Returns how many subscribers the event was handed to. An event published while no one is subscribed is dropped, and the count is 0.

def subscribe(self, /):

Subscribes to the bus, returning an endpoint that sees events published from now on.

def publisher(self, /):

Takes another publisher on the same bus, for another part that announces.