herostools.flow._subscriber

Queued per-source subscriber for flow data.

Attributes

_STOP

Classes

FlowSubscriber

Subscribe to the flow_data event of several HEROs and process messages on one worker.

Module Contents

herostools.flow._subscriber._STOP

class herostools.flow._subscriber.FlowSubscriber(session: Any, realm: str, handler: collections.abc.Callable[[str, Any], None], queue_warn_threshold: int = 100, name: str = 'unnamed')

Subscribe to the flow_data event of several HEROs and process messages on one worker.

The zenoh callback only enqueues the raw bytes. Decoding happens on the worker, so a payload that cannot be decoded (for example one referencing a HERO that is gone) neither blocks zenoh nor kills the worker. Decoding a payload that references a dead HERO builds a RemoteHERO, which blocks the single worker for the discovery timeout (about 2 s) and stalls every source of this subscriber. Control items can be put on the same queue with submit, which keeps message handling and control work strictly ordered (FIFO).

Parameters:

• session – Zenoh session used to declare the subscribers.

• realm – HEROS realm of the observed objects.

• handler – Called as handler(source, payload) on the worker for each decoded message.

• queue_warn_threshold – Queue depth at which a warning is logged. The warning is logged once per crossing and re-armed when the depth falls below the threshold again.

• name – Name of the owning node. It appears in log messages and in the worker thread name flow-subscriber-<name>.

_name = 'unnamed'

_session

_realm

_handler

_queue_warn_threshold = 100

_queue: queue.Queue[Any]

_subscribers: dict[str, Any]

_lock

_warn_lock

_warned = False

_stopped = False

_worker

_key(source: str) → str

add(source: str) → None

Subscribe to the flow_data event of a source. Does nothing if already subscribed.

Parameters:

source – Name of the source HERO.

submit(func: collections.abc.Callable[[], None]) → None

Run a callable on the worker, ordered with the messages.

Parameters:

func – Callable without arguments. Exceptions are logged and do not stop the worker.

stop(timeout: float = 5.0) → None

Undeclare all subscribers and stop the worker. Safe to call more than once.

Queued items are discarded. Only an item already running is finished. If it does not finish within the timeout, a warning is logged and the worker exits afterwards.

Parameters:

timeout – Seconds to wait for the worker to finish.

_on_sample(source: str, sample: Any) → None

Zenoh callback: enqueue the raw bytes without decoding.

_put(item: Any) → None

_run() → None

Worker loop: decode messages, call the handler, run submitted callables.

_dispatch(source: str, raw: bytes) → None

_run_item(func: collections.abc.Callable[[], None]) → None