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]

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

See SubscriberAsync 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.

stop() → None

Stop listening for new events.

Stops the listen() coroutine.

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.

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

Get the next event from the subscriber.

This is a blocking call that waits 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.ZMQPublisherAsync(uri: str, bind: bool = False)

Bases: orangeqs.juice.messaging.protocol.PublisherAsync

Asynchronous client for publishing events to other Juice services over ZMQ.

See PublisherAsync for more information.

async publish(event: orangeqs.juice.messaging.protocol.Event) → None

Publish a message to the system.

Parameters

• event (Event): Event to be published.

class orangeqs.juice.messaging.zmq._pubsub.ZMQPublisherBlocking(uri: str, bind: bool = False)

Bases: orangeqs.juice.messaging.protocol.PublisherBlocking

Synchronous client for publishing events to other Juice services over ZMQ.

See PublisherBlocking for more information.

publish(event: orangeqs.juice.messaging.protocol.Event) → None

Publish a message to the system.

Parameters

• event (Event): Event to be published.

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

Bases: orangeqs.juice.messaging.zmq._pubsub.ZMQSubscriberAsync[orangeqs.juice.messaging.protocol.EventType]

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

See SubscriberAsync for more information.

class orangeqs.juice.messaging.zmq._pubsub.ZMQPublisher(uri: str, bind: bool = False)

Bases: orangeqs.juice.messaging.zmq._pubsub.ZMQPublisherAsync

Asynchronous client for publishing events to other Juice services over ZMQ.

See PublisherAsync for more information.