herostools.flow._subscriber
===========================

.. py:module:: herostools.flow._subscriber

.. autoapi-nested-parse::

   Queued per-source subscriber for flow data.



Attributes
----------

.. autoapisummary::

   herostools.flow._subscriber._STOP


Classes
-------

.. autoapisummary::

   herostools.flow._subscriber.FlowSubscriber


Module Contents
---------------

.. py:data:: _STOP

.. py:class:: 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).

   :param session: Zenoh session used to declare the subscribers.
   :param realm: HEROS realm of the observed objects.
   :param handler: Called as `handler(source, payload)` on the worker for each decoded message.
   :param 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.
   :param name: Name of the owning node. It appears in log messages and in the worker thread name
                `flow-subscriber-<name>`.


   .. py:attribute:: _name
      :value: 'unnamed'



   .. py:attribute:: _session


   .. py:attribute:: _realm


   .. py:attribute:: _handler


   .. py:attribute:: _queue_warn_threshold
      :value: 100



   .. py:attribute:: _queue
      :type:  queue.Queue[Any]


   .. py:attribute:: _subscribers
      :type:  dict[str, Any]


   .. py:attribute:: _lock


   .. py:attribute:: _warn_lock


   .. py:attribute:: _warned
      :value: False



   .. py:attribute:: _stopped
      :value: False



   .. py:attribute:: _worker


   .. py:method:: _key(source: str) -> str


   .. py:method:: add(source: str) -> None

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

      :param source: Name of the source HERO.



   .. py:method:: submit(func: collections.abc.Callable[[], None]) -> None

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

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



   .. py:method:: 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.

      :param timeout: Seconds to wait for the worker to finish.



   .. py:method:: _on_sample(source: str, sample: Any) -> None

      Zenoh callback: enqueue the raw bytes without decoding.



   .. py:method:: _put(item: Any) -> None


   .. py:method:: _run() -> None

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



   .. py:method:: _dispatch(source: str, raw: bytes) -> None


   .. py:method:: _run_item(func: collections.abc.Callable[[], None]) -> None