Class Edge

masfactory/core/edge.py:12–160

Directed message channel between two nodes. An `Edge` buffers at most one in-flight message at a time. It can be opened/closed via a gate, and it becomes *congested* when a message is pending consumption by the receiver.

    from .node import Node

class Edge:
    """Directed message channel between two nodes.

    An `Edge` buffers at most one in-flight message at a time. It can be opened/closed via a
    gate, and it becomes *congested* when a message is pending consumption by the receiver.
    """
    
    class Hook:
        SEND_MESSAGE = 'send_message'
        RECEIVE_MESSAGE = 'receive_message'
    def __init__(self,
            sender:Node,
            receiver:Node,
            keys:dict[str,dict|str]|None=None):
        """
        Args:
            sender: Source node.
            receiver: Target node.
            keys: Message fields forwarded by this edge.
        """
        self._sender:Node = sender
        self._receiver:Node = receiver
        self._keys:dict[str,dict|str] = keys if keys is not None else {"message":""}
        self._message:dict = {}
        self._is_congested:bool = False
        self._gate:Gate = Gate.OPEN
        self._hooks = HookManager()
    @property
    def hooks(self) -> HookManager:
        return self._hooks
    def hook_register(
        self,
        hook_key: object,
        func: Callable,
        recursion: bool = False,
        target_type: type | tuple[type, ...] | None = None,
        target_filter: Callable[[object], bool] | None = None,
        selector: Selector | None = None,
    ) -> None:
        """
        Register edge hooks with selector filtering.

        Edge objects do not support name filtering.

        Args:
            hook_key: Hook stage key.
            func: Hook callback function.
            recursion: Reserved for API parity. Recursion is implemented by graphs.
            target_type: Optional type filter used by the selector.
            target_filter: Optional predicate applied to the matched object.
            selector: Optional explicit selector. When provided, it overrides `target_type/target_filter`.
        """
        predicate = None
        if target_filter is not None:
            predicate = lambda t: t.obj is not None and target_filter(t.obj)

        selector = build_selector(
            selector=selector,