herostools.flow.join
====================

.. py:module:: herostools.flow.join

.. autoapi-nested-parse::

   Flow node that collects one message per input.



Classes
-------

.. autoapisummary::

   herostools.flow.join.Join


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

.. py:class:: Join(inputs: collections.abc.Mapping[str, heros.HERORef | str] | collections.abc.Sequence[heros.HERORef | str], *, max_open_rounds: int = 10, **kwargs)

   Bases: :py:obj:`herostools.flow.node.Node`


   Node that collects one message per input and then processes them together.

   Messages are grouped into rounds by their identifier. `_process` runs once per completed
   round, with the data of every input keyed by input name, in config order. The default
   `_process` emits that dict. The rules:

   - A round completes when every configured input has delivered. The metadata of the inputs is
     merged in config order. Their flow paths are joined in config order without duplicates.
     Completed rounds are processed in completion order.
   - At most `max_open_rounds` incomplete rounds are open. A message with a new identifier at the
     limit drops the oldest incomplete round and logs a warning that names its identifier and
     missing inputs. The warning is logged once per crossing of the limit, and again after the
     number of open rounds fell below the limit.
   - The last `max_open_rounds` finished or dropped identifiers are remembered. Later messages
     for them are ignored.
   - A repeated message from an input for an open round replaces the earlier one.
   - A message that fails validation does not count.
   - `feed` processes its message alone, as input `"manual"`. It never joins a round.

   Rounds are only touched on the worker, so they need no lock.


   .. py:attribute:: _hero_implements
      :type:  ClassVar[list[str]]
      :value: ['herostools.flow.Node', 'herostools.flow.Join']



   .. py:attribute:: _max_open_rounds
      :value: 10



   .. py:attribute:: _rounds
      :type:  dict[Any, dict[str, tuple[Any, dict]]]


   .. py:attribute:: _finished
      :type:  collections.deque


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



   .. py:method:: _process(inputs: dict[str, Any], metadata: dict) -> tuple[Any, dict] | None

      Emit the dict of all inputs. Override in subclasses to combine them.



   .. py:method:: _collect(input_name: str, data: Any, metadata: dict) -> tuple[dict, dict] | None

      Add the message to its round and return the round once every input delivered.



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

      Drop the oldest incomplete round, with a warning once per crossing of the limit.