From c9fda50601bee951707b6e5e60a2af0aaf35942b Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 15 Oct 2026 09:44:22 +0200 Subject: [PATCH] Connection nodes speak of listener and sender, and of send_queue_size MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mqtt and http take websocket's vocabulary: their Mode is LISTENER/SENDER, default names read mqtt_listener_… and http_listener_…, and mqtt's publish_queue_size is send_queue_size. No alias stays behind, so the flow store rewrites the old setting in every published and draft mqtt node when it opens, in the same migration as trigger → debounce. Every mqtt and http setting carries a one-line description for the editor, and the node-types reference uses one wording for direction and says that an http sender publishes a JSON object response on its outputs. --- backend/fluksio/flow/controller.py | 6 +- backend/fluksio/flow/nodes/base.py | 4 +- backend/fluksio/flow/nodes/http.py | 91 ++++++----- backend/fluksio/flow/nodes/mqtt.py | 169 ++++++++++++-------- backend/fluksio/flow/nodes/websocket.py | 8 +- backend/fluksio/flow/store.py | 33 ++-- backend/tests/flow/test_brain_graph.py | 6 +- backend/tests/flow/test_house_vocabulary.py | 8 +- backend/tests/flow/test_senders.py | 28 ++-- backend/tests/flow/test_store.py | 14 ++ docs/getting-started/facility-automation.md | 8 +- docs/reference/node-types.md | 26 +-- 12 files changed, 241 insertions(+), 160 deletions(-) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index ac14ecb..6675701 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -97,7 +97,7 @@ HOOK_PREFIX = "/hooks" def _is_hook(node: Node | None) -> TypeGuard[HttpNode | WebSocketNode]: """Mounted under HOOK_PREFIX: a webhook, or a socket the engine serves.""" - return (isinstance(node, HttpNode) and node.mode is HttpNode.Mode.TRIGGER) or ( + return (isinstance(node, HttpNode) and node.mode is HttpNode.Mode.LISTENER) or ( isinstance(node, WebSocketNode) and node.serves ) @@ -338,13 +338,13 @@ NODE_TYPES: dict[str, NodeType] = { ), "mqtt": NodeType( title="MQTT", - description="Subscribe to topics, or publish what arrives on its inputs.", + description="Listen on topics, or send to them.", cls=MqttNode, params_schema=_schema_of(MqttNode), ), "http": NodeType( title="HTTP", - description="Receive data on a webhook, or send it to a URL.", + description="Listen on a webhook, or send to a URL and pass the reply on.", cls=HttpNode, params_schema=_schema_of(HttpNode), ), diff --git a/backend/fluksio/flow/nodes/base.py b/backend/fluksio/flow/nodes/base.py index 64e22a1..d8fb937 100644 --- a/backend/fluksio/flow/nodes/base.py +++ b/backend/fluksio/flow/nodes/base.py @@ -411,7 +411,7 @@ class Node: Trigger this node externally, executing downstream nodes if dependencies are met. This method is for nodes that receive data from upstream dependencies. - For trigger/subscriber nodes that inject data into the pipeline, use :meth:`inject`. + For listener nodes that inject data into the pipeline, use :meth:`inject`. :param inputs: Input values matching this node's ``requires``. :type inputs: dict | None @@ -429,7 +429,7 @@ class Node: """ Inject data into the pipeline as if this node produced it. - This method is for trigger/subscriber nodes that receive external data + This method is for listener nodes that receive external data (e.g., HTTP requests, MQTT messages) and need to inject it into the pipeline. The data is validated against this node's ``provides`` specification. diff --git a/backend/fluksio/flow/nodes/http.py b/backend/fluksio/flow/nodes/http.py index d9b13c8..9fbf12d 100644 --- a/backend/fluksio/flow/nodes/http.py +++ b/backend/fluksio/flow/nodes/http.py @@ -1,4 +1,4 @@ -"""HTTP nodes: an inbound webhook trigger and an outbound request sender.""" +"""HTTP nodes: an inbound webhook listener and an outbound request sender.""" from __future__ import annotations @@ -81,50 +81,50 @@ def _first_catch_all(app: FastAPI) -> int: class HttpNode(Node): """ - HTTP node that can act as a trigger (receiver) or sender based on configuration. + HTTP node that acts as a listener (webhook) or a sender. This node integrates with FastAPI to either: - - **Trigger mode**: Receive incoming HTTP requests (GET/POST) and inject data + - **Listener**: Receive incoming HTTP requests (GET/POST) and inject data into the pipeline. Used when ``provides`` is specified but ``requires`` is empty. - - **Sender mode**: Make outgoing HTTP requests with pipeline data. Used when - ``requires`` is specified. + - **Sender**: Make outgoing HTTP requests with pipeline data, and publish a + JSON object response on the outputs. Used when ``requires`` is specified. - :param url: The URL endpoint. For trigger mode, this is the route path - (e.g., "/sensors/temperature"). For sender mode, this is the full URL + :param url: The URL endpoint. For a listener, this is the route path + (e.g., "/sensors/temperature"). For a sender, this is the full URL to send requests to. :type url: str :param method: HTTP method - "GET" or "POST". :type method: Literal["GET", "POST"] - :param requires: Messages required by this node (makes it a sender node). + :param requires: Messages required by this node (makes it a sender). :type requires: MessageSpec | list[MessageSpec] | None - :param provides: Messages provided by this node (makes it a trigger node). + :param provides: Messages provided by this node (makes it a listener). :type provides: MessageSpec | list[MessageSpec] | None :param params: Additional parameters for the node. :type params: dict :param name: Optional name for the node. :type name: str | None - :param timeout: Request timeout in seconds (for sender mode). + :param timeout: Request timeout in seconds (sender only). :type timeout: float :param headers: Additional HTTP headers. :type headers: dict[str, str] | None - :param secret: Shared secret callers append to the webhook URL (trigger mode). + :param secret: Shared secret callers append to the webhook URL (listener only). :type secret: str :raises ValueError: If both ``requires`` and ``provides`` are empty, or if the configuration is invalid. :example: - Trigger node (receives POST requests): + Listener (receives POST requests): - >>> trigger = HttpNode( + >>> listener = HttpNode( ... url="/api/sensors/temperature", ... method="POST", ... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)], ... params={}, ... ) - Sender node (makes POST requests): + Sender (makes POST requests): >>> sender = HttpNode( ... url="https://api.example.com/data", @@ -137,7 +137,7 @@ class HttpNode(Node): class Mode(Enum): """Operating mode of the HTTP node.""" - TRIGGER = "trigger" # Receives HTTP requests + LISTENER = "listener" # Receives HTTP requests SENDER = "sender" # Sends HTTP requests # A repeated request is a repeated request, whatever the endpoint does @@ -159,14 +159,29 @@ class HttpNode(Node): class Params(BaseModel): model_config = ConfigDict(extra="allow") - url: str - method: Literal["GET", "POST"] = "POST" - timeout: float = 30.0 - headers: dict[str, str] = {} + url: str = Field( + description=( + "A path such as /tick, served as a webhook at /hooks//, " + "or the full URL a sender requests." + ) + ) + method: Literal["GET", "POST"] = Field( + default="POST", + description="The method a webhook accepts or a sender uses.", + ) + timeout: float = Field( + default=30.0, description="Seconds a sender waits for the response." + ) + headers: dict[str, str] = Field( + default={}, + description=( + "Request headers, sender only. A value may be a secret reference." + ), + ) query: dict[str, Any] = Field( default_factory=dict, description=( - "Fixed query parameters, sender mode. A value may be a secret " + "Fixed query parameters, sender only. A value may be a secret " "reference, which is how an API key stays out of the flow file." ), ) @@ -228,7 +243,7 @@ class HttpNode(Node): ) # Inputs mean this node sends data out; outputs mean it receives. - self.mode = HttpNode.Mode.SENDER if requires else HttpNode.Mode.TRIGGER + self.mode = HttpNode.Mode.SENDER if requires else HttpNode.Mode.LISTENER self.url = cfg.url self.method = cfg.method.upper() @@ -245,12 +260,12 @@ class HttpNode(Node): name = f"http_{self.mode.value}_{safe_url}" # Initialize parent with appropriate function - # For trigger mode, f is a no-op since data is injected via inject() - # For sender mode, f handles the outgoing HTTP request + # For a listener, f is a no-op since data is injected via inject() + # For a sender, f handles the outgoing HTTP request super().__init__( f=( - self._noop_trigger - if self.mode == HttpNode.Mode.TRIGGER + self._noop_listener + if self.mode == HttpNode.Mode.LISTENER else self._sender_handler ), requires=requires, @@ -260,11 +275,11 @@ class HttpNode(Node): ) @staticmethod - def _noop_trigger(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + def _noop_listener(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: """ - No-op function for trigger mode nodes. + No-op function for listener nodes. - Trigger mode nodes inject data via :meth:`inject`, not :meth:`__call__`. + Listener nodes inject data via :meth:`inject`, not :meth:`__call__`. This function exists only to satisfy the Node interface. """ return None @@ -273,7 +288,7 @@ class HttpNode(Node): self, params: dict[str, Any], **kwargs: Any ) -> dict[str, Any] | None: """ - Send HTTP request with pipeline data (sender mode). + Send HTTP request with pipeline data (sender). This method is called when upstream dependencies are satisfied. It sends the required data via HTTP request. @@ -332,37 +347,37 @@ class HttpNode(Node): async def start(self, app: FastAPI | None = None) -> None: """A webhook needs a route; a sender reaches out on its own.""" - if self.mode is HttpNode.Mode.TRIGGER and app is not None: + if self.mode is HttpNode.Mode.LISTENER and app is not None: self.register_route(app) async def stop(self, app: FastAPI | None = None) -> None: - if self.mode is HttpNode.Mode.TRIGGER and app is not None: + if self.mode is HttpNode.Mode.LISTENER and app is not None: self.unregister_route(app) def register_route(self, app: FastAPI) -> None: """ Register this node's HTTP endpoint with a FastAPI application. - This method should only be called for trigger mode nodes. It creates + This method should only be called for listener nodes. It creates a route that, when called, triggers the node in the pipeline. :param app: The FastAPI application instance. :type app: FastAPI - :raises RuntimeError: If called on a sender mode node. + :raises RuntimeError: If called on a sender. :example: >>> from fastapi import FastAPI >>> app = FastAPI() - >>> trigger_node = HttpNode( + >>> listener = HttpNode( ... url="/sensors/data", ... method="POST", ... provides=[MessageSpec(name="value", dtype=DType.FLOAT)], ... params={}, ... ) - >>> trigger_node.register_route(app) + >>> listener.register_route(app) """ - if self.mode != HttpNode.Mode.TRIGGER: - raise RuntimeError("Can only register routes for trigger mode nodes") + if self.mode != HttpNode.Mode.LISTENER: + raise RuntimeError("Can only register routes for listener nodes") if self._route_registered: return @@ -408,7 +423,7 @@ class HttpNode(Node): if spec.port in data: typed_data[spec.port] = spec.coerce(data[spec.port]) - # Inject data into the pipeline (trigger mode nodes inject via + # Inject data into the pipeline (listener nodes inject via # provides). Off the loop: journalling is a blocking Redis # round trip, and every webhook was making it on the thread # the whole API answers from. diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index 7a3303f..a6dbc6b 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/backend/fluksio/flow/nodes/mqtt.py @@ -1,4 +1,4 @@ -"""MQTT nodes: a subscriber that wakes the graph, a publisher that speaks for it.""" +"""MQTT nodes: a listener that wakes the graph, a sender that speaks for it.""" from __future__ import annotations @@ -62,15 +62,14 @@ def topic_matches(filter_: str, topic: str) -> bool: class MqttNode(Node): """ - MQTT node that can act as a subscriber (trigger) or publisher (sender). + MQTT node that acts as a listener or a sender. This node integrates with an MQTT broker to either: - - **Trigger mode (Subscriber)**: Subscribe to MQTT topics and inject received - messages into the pipeline. Used when ``provides`` is specified but - ``requires`` is empty. - - **Sender mode (Publisher)**: Publish pipeline data to MQTT topics. Used when - ``requires`` is specified. + - **Listener**: Subscribe to MQTT topics and inject received messages into + the pipeline. Used when ``provides`` is specified but ``requires`` is empty. + - **Sender**: Publish pipeline data to MQTT topics. Used when ``requires`` + is specified. The ``topic`` parameter in ``params`` controls the mapping between pipeline message names and MQTT topics: @@ -78,11 +77,11 @@ class MqttNode(Node): - **dict**: Explicit mapping from message name to MQTT topic, e.g. ``{"temperature": "sensors/room1/temp", "humidity": "sensors/room1/hum"}``. - **str** (legacy): A single topic string. All messages are mapped to this - one topic (subscriber receives from it, publisher sends to it). + one topic (a listener subscribes to it, a sender publishes to it). - :param requires: Messages required by this node (makes it a publisher node). + :param requires: Messages required by this node (makes it a sender). :type requires: MessageSpec | list[MessageSpec] | None - :param provides: Messages provided by this node (makes it a subscriber node). + :param provides: Messages provided by this node (makes it a listener). :type provides: MessageSpec | list[MessageSpec] | None :param params: Parameters dict containing: - ``topic`` (str | dict): MQTT topic(s). A dict maps message names to @@ -97,7 +96,7 @@ class MqttNode(Node): - ``keepalive`` (int): Keepalive interval in seconds (default: 60) - ``timeout`` (float): Deadline for a broker operation in seconds (default: 10.0) - - ``publish_queue_size`` (int): Publisher backlog depth (default: 256) + - ``send_queue_size`` (int): Sender backlog depth (default: 256) :type params: dict :param name: Optional name for the node. :type name: str | None @@ -105,9 +104,9 @@ class MqttNode(Node): :raises ValueError: If both ``requires`` and ``provides`` are empty. :example: - Subscriber with per-message topics: + Listener with per-message topics: - >>> subscriber = MqttNode( + >>> listener = MqttNode( ... provides=[ ... MessageSpec(name="inverter_input", dtype=DType.FLOAT), ... MessageSpec(name="inverter_output", dtype=DType.FLOAT), @@ -121,9 +120,9 @@ class MqttNode(Node): ... }, ... ) - Publisher with per-message topics: + Sender with per-message topics: - >>> publisher = MqttNode( + >>> sender = MqttNode( ... requires=[ ... MessageSpec(name="target_temp", dtype=DType.FLOAT), ... MessageSpec(name="fan_speed", dtype=DType.INT), @@ -138,9 +137,9 @@ class MqttNode(Node): ... }, ... ) - Legacy single-topic subscriber: + Legacy single-topic listener: - >>> subscriber = MqttNode( + >>> listener = MqttNode( ... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)], ... params={"topic": "sensors/temperature", "broker_host": "localhost"}, ... ) @@ -149,8 +148,8 @@ class MqttNode(Node): class Mode(Enum): """Operating mode of the MQTT node.""" - SUBSCRIBER = "subscriber" # Receives MQTT messages (trigger) - PUBLISHER = "publisher" # Sends MQTT messages (sender) + LISTENER = "listener" # Subscribes and injects what arrives + SENDER = "sender" # Publishes its inputs # Publishing again is a second command to whatever is listening. idempotent = False @@ -167,14 +166,14 @@ class MqttNode(Node): "retain", "keepalive", "timeout", - "publish_queue_size", + "send_queue_size", "json_keys", "_topic_to_ports", "_wildcards", "_subscription_task", "_mqtt_client", "_stop_event", - "_publish_queue", + "_send_queue", "_publisher_task", "_loop", ) @@ -183,15 +182,41 @@ class MqttNode(Node): model_config = ConfigDict(extra="allow") # One topic for every port, or a per-port mapping. - topic: str | dict[str, str] = "*" - broker_host: str = "localhost" - broker_port: int = 1883 - username: str | None = None - password: str | None = Field(default=None, json_schema_extra={"x-secret": True}) - client_id: str | None = None - qos: int = 0 - retain: bool = False - keepalive: int = 60 + topic: str | dict[str, str] = Field( + default="*", + description='One topic for every port, or {"port": "some/topic"} per port.', + ) + broker_host: str = Field( + default="localhost", + description="Hostname or IP of the broker, as the engine reaches it.", + ) + broker_port: int = Field( + default=1883, description="The port the broker listens on." + ) + username: str | None = Field( + default=None, description="Username for the broker, if it asks for one." + ) + password: str | None = Field( + default=None, + description="Password for the broker, kept in the secrets store.", + json_schema_extra={"x-secret": True}, + ) + client_id: str | None = Field( + default=None, + description="Client identifier the broker sees. Empty picks a random one.", + ) + qos: int = Field( + default=0, + description="Quality of service for subscribing and publishing, 0, 1 or 2.", + ) + retain: bool = Field( + default=False, + description="Have the broker keep the last value sent for later subscribers.", + ) + keepalive: int = Field( + default=60, + description="Seconds between pings that keep an idle connection open.", + ) # Bounds every broker operation: subscribe, publish, and the # disconnect acknowledgement on the way out. Without one a client # whose socket died waits for that ack forever, and the task never @@ -202,9 +227,9 @@ class MqttNode(Node): description="Give up on a broker operation after this many seconds.", ) # Deep enough to ride out a broker hiccup, shallow enough that a - # publisher which cannot keep up drops old values instead of growing + # sender which cannot keep up drops old values instead of growing # without bound. A node that bursts wants more than one that trickles. - publish_queue_size: int = Field( + send_queue_size: int = Field( default=256, gt=0, description=( @@ -215,17 +240,23 @@ class MqttNode(Node): # Which key to lift out of a JSON object payload. A device that wraps # its reading — Victron's ``{"value": 5}`` — is otherwise a Python node # per port. One key for every port, or a per-port mapping. - json_key: str | dict[str, str] = "" + json_key: str | dict[str, str] = Field( + default="", + description=( + "Key to lift out of a JSON object payload, for every port or as " + '{"port": "key"}.' + ), + ) @classmethod def target_key(cls, params: dict[str, Any]) -> str | None: """The broker and topic, which is one physical thing. - A publisher and a subscriber on the same topic get the same key on + A sender and a listener on the same topic get the same key on purpose: they are two ends of one wire, and drawing them as one neuron is the only way the path through the broker shows up at all. """ - # ponytail: publisher and subscriber merge into one neuron; key on mode + # ponytail: sender and listener merge into one neuron; key on mode # as well if the two directions ever need telling apart. fields = cls.Params.model_fields topic = params.get("topic", fields["topic"].default) @@ -253,9 +284,9 @@ class MqttNode(Node): ) # Inputs mean this node publishes; outputs mean it subscribes. - self.mode = MqttNode.Mode.PUBLISHER if requires else MqttNode.Mode.SUBSCRIBER + self.mode = MqttNode.Mode.SENDER if requires else MqttNode.Mode.LISTENER - ports = provides if self.mode == MqttNode.Mode.SUBSCRIBER else requires + ports = provides if self.mode == MqttNode.Mode.LISTENER else requires if isinstance(cfg.topic, dict): self.topics: dict[str, str] = dict(cfg.topic) else: @@ -285,13 +316,13 @@ class MqttNode(Node): self.retain = cfg.retain self.keepalive = cfg.keepalive self.timeout = cfg.timeout - self.publish_queue_size = cfg.publish_queue_size + self.send_queue_size = cfg.send_queue_size # Runtime state self._subscription_task: asyncio.Task[None] | None = None self._mqtt_client = None self._stop_event: asyncio.Event | None = None - self._publish_queue: asyncio.Queue[dict[str, Any]] | None = None + self._send_queue: asyncio.Queue[dict[str, Any]] | None = None self._publisher_task: asyncio.Task[None] | None = None self._loop: asyncio.AbstractEventLoop | None = None @@ -311,13 +342,13 @@ class MqttNode(Node): name = f"mqtt_{self.mode.value}_{safe_topic}" # Initialize parent with appropriate function - # For subscriber mode, f is a no-op since data is injected via inject() - # For publisher mode, f handles the outgoing MQTT publish + # For a listener, f is a no-op since data is injected via inject() + # For a sender, f handles the outgoing MQTT publish super().__init__( f=( - self._noop_subscriber - if self.mode == MqttNode.Mode.SUBSCRIBER - else self._publisher_handler + self._noop_listener + if self.mode == MqttNode.Mode.LISTENER + else self._sender_handler ), requires=requires, provides=provides, @@ -326,13 +357,11 @@ class MqttNode(Node): ) @staticmethod - def _noop_subscriber( - params: dict[str, Any], **kwargs: Any - ) -> dict[str, Any] | None: + def _noop_listener(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: """ - No-op function for subscriber mode nodes. + No-op function for listener nodes. - Subscriber mode nodes inject data via :meth:`inject`, not :meth:`__call__`. + Listener nodes inject data via :meth:`inject`, not :meth:`__call__`. This function exists only to satisfy the Node interface and should not be called directly. @@ -345,11 +374,11 @@ class MqttNode(Node): """ return None - def _publisher_handler( + def _sender_handler( self, params: dict[str, Any], **kwargs: Any ) -> dict[str, Any] | None: """ - Publish pipeline data to MQTT topic (publisher mode). + Publish pipeline data to MQTT topic (sender). This method is called when upstream dependencies are satisfied. Handing the payload to the node's publisher task is all that happens @@ -363,7 +392,7 @@ class MqttNode(Node): :returns: None (publishing is fire-and-forget). :rtype: dict | None """ - loop, queue = self._loop, self._publish_queue + loop, queue = self._loop, self._send_queue if loop is not None and queue is not None: loop.call_soon_threadsafe(self._enqueue, queue, dict(kwargs)) return None @@ -386,21 +415,21 @@ class MqttNode(Node): if queue.full(): try: queue.get_nowait() - logger.warning("Publish queue full for node '%s', dropped", self.name) - self.report_health("degraded", "publish queue full") + logger.warning("Send queue full for node '%s', dropped", self.name) + self.report_health("degraded", "send queue full") except asyncio.QueueEmpty: pass queue.put_nowait(data) async def _publisher_loop(self) -> None: - """Hold one connection and drain the publish queue over it. + """Hold one connection and drain the send queue over it. A dropped connection raises, and the supervisor decides when to - reconnect — the same arrangement the subscriber uses. + reconnect — the same arrangement the listener uses. """ aiomqtt = _aiomqtt() - queue = self._publish_queue + queue = self._send_queue if queue is None: return @@ -461,7 +490,7 @@ class MqttNode(Node): continue # A string goes on the wire as it stands. Devices on a shared - # broker expect bare values, and the subscriber below already + # broker expect bare values, and the listener below already # falls back to the raw text when it is not JSON, so a # fluksio-to-fluksio round trip is unaffected. payload = value if isinstance(value, str) else json.dumps(value) @@ -479,8 +508,8 @@ class MqttNode(Node): ) async def start(self, app: FastAPI | None = None) -> None: - """A subscriber listens; a publisher opens the connection it will reuse.""" - if self.mode is MqttNode.Mode.SUBSCRIBER: + """A listener subscribes; a sender opens the connection it will reuse.""" + if self.mode is MqttNode.Mode.LISTENER: await self.start_subscription() else: await self.start_publisher() @@ -491,41 +520,41 @@ class MqttNode(Node): async def start_publisher(self) -> None: """Run the task that owns this node's connection to the broker.""" - if self._publish_queue is not None: + if self._send_queue is not None: return - self._publish_queue = asyncio.Queue(maxsize=self.publish_queue_size) + self._send_queue = asyncio.Queue(maxsize=self.send_queue_size) self._loop = asyncio.get_running_loop() self._publisher_task = self._run_supervised("mqtt-out", self._publisher_loop) async def stop_publisher(self) -> None: """Drop the queue and let the connection go.""" - if self._publish_queue is None: + if self._send_queue is None: return if self._publisher_task is not None: await self._cancel_task(self._publisher_task) self._publisher_task = None - self._publish_queue = None + self._send_queue = None self._loop = None async def start_subscription(self) -> None: """ - Start the MQTT subscription for trigger mode nodes. + Start the MQTT subscription for listener nodes. This method starts a background task that listens for messages on the subscribed topic and triggers the pipeline when messages arrive. - :raises RuntimeError: If called on a publisher mode node. + :raises RuntimeError: If called on a sender. :example: - >>> subscriber = MqttNode( + >>> listener = MqttNode( ... topic="sensors/#", ... provides=[MessageSpec(name="value", dtype=DType.FLOAT)], ... params={"broker_host": "localhost"}, ... ) - >>> await subscriber.start_subscription() + >>> await listener.start_subscription() """ - if self.mode != MqttNode.Mode.SUBSCRIBER: - raise RuntimeError("Can only start subscription for subscriber mode nodes") + if self.mode != MqttNode.Mode.LISTENER: + raise RuntimeError("Can only start subscription for listener nodes") if self._subscription_task is not None: return # Already running diff --git a/backend/fluksio/flow/nodes/websocket.py b/backend/fluksio/flow/nodes/websocket.py index 15732f6..6593758 100644 --- a/backend/fluksio/flow/nodes/websocket.py +++ b/backend/fluksio/flow/nodes/websocket.py @@ -121,7 +121,13 @@ class WebSocketNode(Node): ), ) # Same as the mqtt node: a device that wraps its reading in an object. - json_key: str | dict[str, str] = "" + json_key: str | dict[str, str] = Field( + default="", + description=( + "Key to lift out of a JSON object frame, for every port or as " + '{"port": "key"}.' + ), + ) @classmethod def target_key(cls, params: dict[str, Any]) -> str | None: diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index 6d2a48f..aee1d9b 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -169,15 +169,16 @@ class FlowStore: self._migrate() def _migrate(self) -> None: - """Rewrite node types stored under a name they no longer have. + """Rewrite node types and settings stored under a name they no longer have. - The node type ``trigger`` became ``debounce`` on 2026-10-15, with no - alias left for the old name. Each published and draft document is read - as plain JSON and only the ``type`` of a matching node changes, so a - field this version does not model survives, and the old document stays - in git. The version goes up by one, so an editor still holding the old - document gets a conflict on its next save instead of writing the old - name back. A store with nothing to rename is left alone. + On 2026-10-15 the node type ``trigger`` became ``debounce`` and the mqtt + setting ``publish_queue_size`` became ``send_queue_size``, with no alias + left for the old names. Each published and draft document is read as + plain JSON and only the matching node changes, so a field this version + does not model survives, and the old document stays in git. The version + goes up by one, so an editor still holding the old document gets a + conflict on its next save instead of writing the old name back. A store + with nothing to rename is left alone. """ changed = [] paths = [*self.root.glob("*/flow.json"), *self.root.glob("*/flow.draft.json")] @@ -185,21 +186,31 @@ class FlowStore: try: data = json.loads(path.read_text()) nodes = [n for n in data["nodes"] if n.get("type") == "trigger"] + queues = [ + n["params"] + for n in data["nodes"] + if n.get("type") == "mqtt" + and "publish_queue_size" in (n.get("params") or {}) + ] except Exception: # Unreadable to the engine too, which reports it when it loads. continue - if not nodes: + if not nodes and not queues: continue for node in nodes: node["type"] = "debounce" + for params in queues: + params["send_queue_size"] = params.pop("publish_queue_size") data["version"] = data.get("version", 1) + 1 path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") changed.append(f"{path.parent.name}/{path.name}") if changed: logger.warning( - "Renamed node type 'trigger' to 'debounce' in %s", ", ".join(changed) + "Renamed node type 'trigger' to 'debounce' and mqtt " + "'publish_queue_size' to 'send_queue_size' in %s", + ", ".join(changed), ) - self._commit("Rename node type 'trigger' to 'debounce'") + self._commit("Rename 'trigger' nodes and mqtt 'publish_queue_size'") # ------------------------------------------------------------------------- # git diff --git a/backend/tests/flow/test_brain_graph.py b/backend/tests/flow/test_brain_graph.py index 72876d7..ec20a45 100644 --- a/backend/tests/flow/test_brain_graph.py +++ b/backend/tests/flow/test_brain_graph.py @@ -19,7 +19,7 @@ from fluksio.flow.store import FlowStore BROKER = {"broker_host": "mosquitto", "broker_port": 1883, "topic": "sensors/temp"} -def subscriber(node_id: str) -> NodeDef: +def listener(node_id: str) -> NodeDef: return NodeDef( id=node_id, type="mqtt", @@ -35,7 +35,7 @@ def controller(tmp_path: Path) -> FlowController: FlowDef( name="house", nodes=[ - subscriber("sensor"), + listener("sensor"), NodeDef( id="scale", type="change", @@ -45,7 +45,7 @@ def controller(tmp_path: Path) -> FlowController: ], ) ) - store.write_flow(FlowDef(name="shed", nodes=[subscriber("sensor")])) + store.write_flow(FlowDef(name="shed", nodes=[listener("sensor")])) return FlowController(store) diff --git a/backend/tests/flow/test_house_vocabulary.py b/backend/tests/flow/test_house_vocabulary.py index 8df6fbb..aff89a6 100644 --- a/backend/tests/flow/test_house_vocabulary.py +++ b/backend/tests/flow/test_house_vocabulary.py @@ -29,7 +29,7 @@ def test_a_filter_does_not_cover_a_neighbouring_topic(): assert not topic_matches("sensors/living/temp", "sensors/living") -def _subscriber(topic, **params) -> MqttNode: +def _listener(topic, **params) -> MqttNode: node = MqttNode( provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)], params={"topic": {"reading": topic}, **params}, @@ -40,7 +40,7 @@ def _subscriber(topic, **params) -> MqttNode: def test_a_wildcard_subscription_routes_to_its_port(): """It used to subscribe and then drop every message it was sent.""" - node = _subscriber("shellypm/status/#") + node = _listener("shellypm/status/#") assert node._ports_for("shellypm/status/pm1:0") == ["reading"] assert node._ports_for("shellypmc/status/pm1:0") == [] @@ -63,12 +63,12 @@ def test_an_exact_topic_still_wins_over_a_filter(): def test_json_key_lifts_the_value_a_device_wraps(): """Victron publishes {"value": 47} on every one of its topics.""" - node = _subscriber("N/x/battery/Soc", json_key="value") + node = _listener("N/x/battery/Soc", json_key="value") assert node.json_keys == {"reading": "value"} def test_without_json_key_the_port_name_is_the_key_as_before(): - node = _subscriber("sensors/temp") + node = _listener("sensors/temp") assert node.json_keys == {} diff --git a/backend/tests/flow/test_senders.py b/backend/tests/flow/test_senders.py index 655066a..1cbfbb8 100644 --- a/backend/tests/flow/test_senders.py +++ b/backend/tests/flow/test_senders.py @@ -20,7 +20,7 @@ def test_http_senders_share_one_pooled_client(): close_shared_client() -def _publisher() -> MqttNode: +def _sender() -> MqttNode: node = MqttNode( requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)], params={"topic": {"setpoint": "heating/setpoint"}}, @@ -29,27 +29,27 @@ def _publisher() -> MqttNode: return node -def test_a_started_publisher_queues_instead_of_connecting(): +def test_a_started_sender_queues_instead_of_connecting(): """The handler runs on a worker thread; it must not block on the broker.""" - node = _publisher() + node = _sender() async def scenario() -> None: - node._publish_queue = asyncio.Queue(maxsize=4) + node._send_queue = asyncio.Queue(maxsize=4) node._loop = asyncio.get_running_loop() - await asyncio.to_thread(node._publisher_handler, {}, setpoint=21.0) + await asyncio.to_thread(node._sender_handler, {}, setpoint=21.0) # call_soon_threadsafe lands on the next loop pass. await asyncio.sleep(0) - assert node._publish_queue.qsize() == 1 - assert node._publish_queue.get_nowait() == {"setpoint": 21.0} + assert node._send_queue.qsize() == 1 + assert node._send_queue.get_nowait() == {"setpoint": 21.0} asyncio.run(scenario()) -def test_a_full_publish_queue_drops_the_oldest(): +def test_a_full_send_queue_drops_the_oldest(): """A broker that cannot keep up must not grow the queue without bound.""" - node = _publisher() + node = _sender() health: list[tuple[str, str | None]] = [] node._on_health = lambda _n, status, detail: health.append((status, detail)) @@ -61,7 +61,7 @@ def test_a_full_publish_queue_drops_the_oldest(): assert queue.qsize() == 2 assert queue.get_nowait() == {"setpoint": 2.0} assert queue.get_nowait() == {"setpoint": 3.0} - assert health == [("degraded", "publish queue full")] + assert health == [("degraded", "send queue full")] asyncio.run(scenario()) @@ -121,7 +121,7 @@ def test_the_configured_timeout_reaches_the_broker_client(monkeypatch): assert seen["timeout"] == 2.5 -def test_the_configured_backlog_reaches_the_publish_queue(monkeypatch): +def test_the_configured_backlog_reaches_the_send_queue(monkeypatch): """The depth is read when the queue is built, so it has to be per node.""" import aiomqtt @@ -139,14 +139,14 @@ def test_the_configured_backlog_reaches_the_publish_queue(monkeypatch): node = MqttNode( requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)], - params={"topic": {"setpoint": "heating/setpoint"}, "publish_queue_size": 8}, + params={"topic": {"setpoint": "heating/setpoint"}, "send_queue_size": 8}, ) node.assign_flow("heating", "out") async def scenario() -> int: await node.start_publisher() - assert node._publish_queue is not None - size = node._publish_queue.maxsize + assert node._send_queue is not None + size = node._send_queue.maxsize await node.stop_publisher() return size diff --git a/backend/tests/flow/test_store.py b/backend/tests/flow/test_store.py index 13abd34..a5e37d3 100644 --- a/backend/tests/flow/test_store.py +++ b/backend/tests/flow/test_store.py @@ -159,3 +159,17 @@ def test_opening_the_store_renames_trigger_nodes_to_debounce(store: FlowStore): FlowStore(store.root) assert commit_count(store) == commits + 1 + + +def test_opening_the_store_renames_the_mqtt_publish_queue_size(store: FlowStore): + """The mqtt setting is now `send_queue_size`, with no alias for the old name.""" + relay = NodeDef(id="relay", type="mqtt", params={"publish_queue_size": 8}) + store.write_flow(FlowDef(name="hall", nodes=[relay])) + commits = commit_count(store) + + reopened = FlowStore(store.root) + + published = reopened.read_flow("hall") + assert published.nodes[0].params == {"send_queue_size": 8} + assert published.version == 2 + assert commit_count(store) == commits + 1 diff --git a/docs/getting-started/facility-automation.md b/docs/getting-started/facility-automation.md index e0b850b..ac5a02f 100644 --- a/docs/getting-started/facility-automation.md +++ b/docs/getting-started/facility-automation.md @@ -86,10 +86,10 @@ Publish the change. The Function now responds to sensor readings instead of the test value. If your sensor sends a JSON object such as `{"temperature": 18.0}`, set `json_key` to `temperature`. -An MQTT node with input ports publishes to its configured topic; one with -output ports subscribes. Add a publishing node only after checking the payload -format your device expects. A relay may need `ON` and `OFF` rather than a -boolean. +An MQTT node with inputs is a sender and publishes to its configured topic; +one with outputs only is a listener and subscribes. Add a sender only after +checking the payload format your device expects. A relay may need `ON` and +`OFF` rather than a boolean. !!! tip "Keep the first test on screen" diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index 9300fed..0157174 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -25,9 +25,11 @@ arrive as keyword arguments. ### MQTT -**`mqtt`** — subscribe to topics, or publish what arrives on its inputs. +**`mqtt`** — listen on topics, or send to them. -A node with *outputs only* subscribes; a node with *inputs* publishes. +The ports pick the direction: outputs only make a **listener**, inputs a +**sender**. A listener subscribes to its topics, a sender publishes its inputs +to them. | Setting | Default | Notes | |---|---|---| @@ -40,7 +42,7 @@ A node with *outputs only* subscribes; a node with *inputs* publishes. | `retain` | `false` | on published messages | | `keepalive` | `60` | seconds | | `timeout` | `10.0` | seconds before a broker operation is given up on | -| `publish_queue_size` | `256` | payloads that may wait for the broker; past this the oldest is dropped and the node reports degraded | +| `send_queue_size` | `256` | payloads that may wait for the broker; past this the oldest is dropped and the node reports degraded | | `json_key` | — | key to lift out of an object payload; one for every port, or `{"port": "key"}` | A topic may be a filter: `+` matches one level, `#` the rest. Everything a @@ -55,18 +57,22 @@ unwrapped only if it contains the port's name as a key. ### HTTP -**`http`** — receive data on a webhook, or send it to a URL. +**`http`** — listen on a webhook, or send to a URL and pass the reply on. -Outputs only makes it a **webhook**: the engine mounts a route at -`/hooks/{flow}/{url}/{secret}` while the flow runs. Inputs make it a **sender**. +The ports pick the direction: outputs only make a **listener**, inputs a +**sender**. A listener is a **webhook**: the engine mounts a route at +`/hooks/{flow}/{url}/{secret}` while the flow runs. A sender requests `url` +and publishes a JSON object response on its outputs, one key per output port. +A key no port declares fails the call, and a response that is not an object +publishes nothing. | Setting | Default | Notes | |---|---|---| -| `url` | — | the route path in webhook mode, the full URL in sender mode | +| `url` | — | the route path for a webhook, the full URL for a sender | | `method` | `POST` | `GET` or `POST` | -| `timeout` | `30` | seconds, sender mode | -| `headers` | `{}` | | -| `query` | `{}` | fixed query parameters; a value may be a secret reference | +| `timeout` | `30` | seconds, sender only | +| `headers` | `{}` | request headers, sender only; a value may be a secret reference | +| `query` | `{}` | fixed query parameters, sender only; a value may be a secret reference | | `send_inputs` | `true` | off when the inputs only trigger the request | | `secret` | — | shared secret appended to the webhook URL; takes a secret reference |