ezmsg.qt.gate#

Gate unit for controlling message flow in processor graphs.

Classes

class GateMessage(open)[source]#

Bases: object

Control message for opening/closing a gate.

Parameters:: open (bool)

open: bool#

__init__(open)#

Parameters:: open (bool)
Return type:: None

class MessageGate(*args, settings=None, **kwargs)[source]#

Bases: Unit

Gate unit that passes or blocks messages based on control input.

When the gate is open, messages flow through. When closed, messages are dropped. Gate state is controlled via INPUT_GATE stream.

Parameters:: settings (Settings | None)

SETTINGS#: alias of MessageGateSettings

STATE#: alias of MessageGateState

INPUT = InputStream:unlocated[Any]()#

OUTPUT = OutputStream:unlocated[Any](self.num_buffers=32, self.force_tcp=None, self.allow_local=None)#

INPUT_GATE = InputStream:unlocated[GateMessage]()#

async initialize()[source]#

Runs when the Unit is instantiated.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should initialize your unit’s state and prepare for message processing.

Return type:: None

async on_gate(msg)[source]#

Handle gate control messages.

Return type:: None
Parameters:: msg (GateMessage)

async on_message(msg)[source]#

Pass message through if gate is open.

Return type:: AsyncGenerator
Parameters:: msg (Any)

class MessageGateSettings(start_open=True)[source]#

Bases: Settings

Settings for MessageGate unit.

Parameters:: start_open (bool)

start_open: bool = True#

__init__(start_open=True)#

Parameters:: start_open (bool)
Return type:: None

class MessageGateState[source]#

Bases: State

Runtime state for MessageGate unit.

is_open: bool#