orxtra v0.13.0 /trace.src.orxtra.trace._pg_event_bus
Updated
On this page

PgEventBus implements EventBus via asyncpg LISTEN/NOTIFY, sharing one connection per channel across all subscribed callbacks and releasing it once the last callback unsubscribes.

#trace.src.orxtra.trace._pg_event_bus

#trace.src.orxtra.trace._pg_event_bus

#_ChannelState

Per-channel state: the asyncpg connection, notification handler, and list of application-level callbacks.

#PgEventBus

PostgreSQL LISTEN/NOTIFY implementation of EventBus.

Wraps asyncpg LISTEN/NOTIFY into the EventBus protocol. Multiple callbacks can subscribe to the same channel; a single asyncpg LISTEN connection is shared across all callbacks for a channel.

#subscribe

python
async def subscribe(self, channel: str, callback: Callable[[str], Awaitable[None]]) -> None

Subscribe a callback to a channel.

The first callback on a channel acquires a connection and registers the asyncpg LISTEN. Subsequent callbacks on the same channel are appended without additional connections.

#unsubscribe

python
async def unsubscribe(self, channel: str, callback: Callable[[str], Awaitable[None]]) -> None

Remove a specific callback from a channel by identity.

When the last callback is removed, the asyncpg LISTEN is cancelled and the connection is released back to the pool.

#publish

python
async def publish(self, channel: str, payload: str) -> None

Publish a notification on a channel via NOTIFY.

#close

python
async def close(self) -> None

Unsubscribe from all channels and release connections.

Search