herostools.flow.join

Flow node that collects one message per input.

Classes

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

Module Contents

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

Bases: 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.

_hero_implements: ClassVar[list[str]] = ['herostools.flow.Node', 'herostools.flow.Join']

_max_open_rounds = 10

_rounds: dict[Any, dict[str, tuple[Any, dict]]]

_finished: collections.deque

_warned = False

_process(inputs: dict[str, Any], metadata: dict) → tuple[Any, dict] | None

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

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

_drop_oldest_round() → None

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