`orangeqs.juice.messaging.zmq._pubsub`#

ZMQ implementation of publisher and subscriber.

Module Contents#

Classes#

`ZMQSubscriberAsync`	Asynchronous client for listening to events from other Juice services over ZMQ.
`ZMQSubscriberBlocking`	Synchronous client for listening to events from other Juice services over ZMQ.
`ZMQPublisherAsync`	Asynchronous client for publishing events to other Juice services over ZMQ.
`ZMQPublisherBlocking`	Synchronous client for publishing events to other Juice services over ZMQ.
`ZMQSubscriber`	Asynchronous client for listening to events from other Juice services over ZMQ.
`ZMQPublisher`	Asynchronous client for publishing events to other Juice services over ZMQ.

API#

class orangeqs.juice.messaging.zmq._pubsub.ZMQSubscriberAsync(uri: str, *, queue: asyncio.Queue[orangeqs.juice.messaging.protocol.EventType] | None = None)#

Bases: orangeqs.juice.messaging.zmq._pubsub._SubscriptionRegistry[orangeqs.juice.messaging.protocol.EventType], orangeqs.juice.messaging.protocol.SubscriberAsync[orangeqs.juice.messaging.protocol.EventType], typing.Generic[orangeqs.juice.messaging.protocol.EventType]

Asynchronous client for listening to events from other Juice services over ZMQ.

See SubscriberAsync for more information.

property queue: asyncio.Queue[orangeqs.juice.messaging.protocol.EventType]#

Queue to which new messages will be put that are received by listen().

Deprecated, use the get() method to retrieve messages instead.

subscribe(event_type: type[orangeqs.juice.messaging.protocol.EventType], *, topic: str | None = None) → None#

Subscribe to an event.

It is possible to subscribe to multiple event types and/or topics. The subscriber allows filtering by event type and an optional topic. The topic filter checks if the topic of the event starts with the topic filter. This means that a topic filter with value "a.b" will match events with topic "a.b" and "a.b.c", but not "a.c" or "a.bb". Note that each topic filter always targets a specific event type, thus they are not shared between different event types.

Parameters#

event_type (type[EventType]): Type of event to subscribe to. Will subscribe only to one specific event type, so not types that inherit this type.
topic (str, optional): Optional topic to filter events. See above for explanation. If not provided will subscribe to all events of specified type.

stop() → None#

Stop listening for new events.

Stops the listen() coroutine.

Deprecated, use the get() method to retrieve messages instead.

async listen() → None#

Coroutine to listen for new events and append them to the queue.

Runs continuously until .stop() is called. This coroutine should be continuously running to receive new messages and append them to the queue.

Deprecated, use the get() method to retrieve messages instead.

async get(timeout: float | None = None) → orangeqs.juice.messaging.protocol.EventType#

Get the next event from the subscriber.

Waits asynchronously until a new event is received.

Parameters#

timeout (float, optional): Optional timeout in seconds to wait for a new event. If not provided, will wait indefinitely.

Raises#

(TimeoutError): If a timeout is provided and no event is received within the timeout period.

class orangeqs.juice.messaging.zmq._pubsub.ZMQSubscriberBlocking(uri: str)#

Bases: orangeqs.juice.messaging.zmq._pubsub._SubscriptionRegistry[orangeqs.juice.messaging.protocol.EventType], orangeqs.juice.messaging.protocol.SubscriberBlocking[orangeqs.juice.messaging.protocol.EventType]

Synchronous client for listening to events from other Juice services over ZMQ.

See SubscriberBlocking for more information.

subscribe(event_type: type[orangeqs.juice.messaging.protocol.EventType], *, topic: str | None = None) → None#

Subscribe to an event.

It is possible to subscribe to multiple event types and/or topics. The subscriber allows filtering by event type and an optional topic. The topic filter checks if the topic of the event starts with the topic filter. This means that a topic filter with value "a.b" will match events with topic "a.b" and "a.b.c", but not "a.c" or "a.bb". Note that each topic filter always targets a specific event type, thus they are not shared between different event types.

Parameters#

event_type (type[EventType]): Type of event to subscribe to. Will subscribe only to one specific event type, so not types that inherit this type.
topic (str, optional): Optional topic to filter events. See above for explanation. If not provided will subscribe to all events of specified type.

get(timeout: float | None = None) → orangeqs.juice.messaging.protocol.EventType#