orangeqs.juice.service._remote#

Base class for remote services.

Module Contents#

Classes#

RemoteService

Base class a remote service is built on.

RemotePublisherAsync

A :class:~orangeqs.juice.messaging.protocol.PublisherAsync over /hub/pub.

RemoteSubscriberAsync

A :class:~orangeqs.juice.messaging.protocol.SubscriberAsync over /hub/sub.

API#

exception orangeqs.juice.service._remote.RemoteAcknowledgementError#

Bases: RuntimeError

Raised when Juice rejects the remote at connection time.

class orangeqs.juice.service._remote.RemoteService(name: str, edge_url: str, api_token: str)#

Bases: orangeqs.juice.service._task_server.TaskHandlerRegistry

Base class a remote service is built on.

Parameters#

  • name (str, optional): The remote’s routing name; must match a config-declared remote.

  • edge_url (str, optional): Base WebSocket URL to reach Juice, e.g. ws://juice.example.com:8888.

  • api_token (str): The edge token the remote presents to Juice. Generate it on the token page of your installation (/hub/token).

run() → None#

Connect to the edge and serve tasks until the channel closes.

async serve() → None#

Open the Reverse Channel, announce, then dispatch pushed tasks.

publisher_async() → orangeqs.juice.service._remote.RemotePublisherAsync#

Return a publisher that emits events onto bus via /hub/pub.

subscriber_async(*, queue: asyncio.Queue[orangeqs.juice.messaging.protocol.Event] | None = None) → orangeqs.juice.service._remote.RemoteSubscriberAsync[orangeqs.juice.messaging.protocol.Event]#

Return a subscriber that streams bus events from /hub/sub.

class orangeqs.juice.service._remote.RemotePublisherAsync(edge_url: str, api_token: str)#

Bases: orangeqs.juice.messaging.protocol.PublisherAsync

A :class:~orangeqs.juice.messaging.protocol.PublisherAsync over /hub/pub.

TODO: a WS handshake per publish. Fine for the low event rate a remote produces; hold a persistent /hub/pub socket if a remote ever becomes a hot publisher.

Parameters#

  • edge_url (str): Base WebSocket URL to reach Juice, e.g. ws://juice.example.com:8888.

  • api_token (str): The edge token the remote presents to Juice.

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

Publish a message to the system.

Parameters#

  • event (Event): Event to be published.

async send_frame(frame: list[bytes]) → None#

Publish a serialized frame to the system.

Parameters#

  • frame (list[bytes]): Serialized frame to be published.

class orangeqs.juice.service._remote.RemoteSubscriberAsync(edge_url: str, api_token: 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]

A :class:~orangeqs.juice.messaging.protocol.SubscriberAsync over /hub/sub.

Parameters#

  • edge_url (str): Base WebSocket URL to reach Juice, e.g. ws://juice.example.com:8888.

  • api_token (str): The edge token the remote presents to Juice.

  • queue (asyncio.Queue, optional): Deprecated; retained only for interface parity with in-Juice subscribers.

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.

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.

close() → None#

Close the subscription WebSocket, if one was opened.

abstract 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.

abstractmethod stop() → None#

Stop listening for new events.

Stops the listen() coroutine.

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

abstractmethod 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.