loop

class EventloopMixin

Mixin to provide common asyncio zmq scaffolding to networked classes.

Inheriting classes should, in order

• call the _init_loop method to create the eventloop, context, and poller

• populate the private _sockets and _receivers dicts

• await the _poll_sockets method, which polls indefinitely.

Inheriting classes must ensure that _init_loop is called in the thread it is intended to run in, and that thread must already have a running eventloop. asyncio eventloops (and most of asyncio) are not thread safe.

To help avoid cross-threading issues, the context() and loop() properties do not automatically create the objects, raising a RuntimeError if they are accessed before _init_loop is called.

To handle slow subscribers in PUB/SUB, use an XPUB instead, and each subscriber also subscribes to a topic with the SUBSCRIBER_PREFIX:id. The XPUB then polls for receive events with _poll_subscriptions, and maintains a list of nodes that have subscribed to it. Subscribers can then wait for the publisher to acknowledge its subscription before marking themselves as ready.

SUBSCRIBER_PREFIX: ClassVar[str] = '__subscriber__:'

property context: Context

property loop: AbstractEventLoop

property sockets: dict[str, Socket]

register_socket(name: str, socket: Socket, receiver: bool = False) → None

Register a socket, optionally declaring it as a receiver socket to poll

add_callback(socket: str, callback: Callable[[Message], Any] | Callable[[Message], Coroutine]) → None

Add a callback to be called when the socket receives a message. Callbacks are called in the order in which they are added.

clear_callbacks() → None