Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Connection nodes speak of listener and sender, and of send_queue_size

Browse Source 
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.
This commit is contained in:
stroblme
2026-10-15 09:44:22 +02:00
parent c032898135
commit c9fda50601
12 changed files with 241 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/flow/controller.py
+3 -3
View File
@@ -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),
),
backend/fluksio/flow/nodes/base.py
+2 -2
View File
@@ -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.
backend/fluksio/flow/nodes/http.py
+53 -38
View File
@@ -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/<flow>/<path>, "
"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.
backend/fluksio/flow/nodes/mqtt.py
+99 -70
View File
@@ -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
backend/fluksio/flow/nodes/websocket.py
+7 -1
View File
@@ -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:
backend/fluksio/flow/store.py
+22 -11
View File
@@ -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
backend/tests/flow/test_brain_graph.py
+3 -3
View File
@@ -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)
backend/tests/flow/test_house_vocabulary.py
+4 -4
View File
@@ -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 == {}
backend/tests/flow/test_senders.py
+14 -14
View File
@@ -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
backend/tests/flow/test_store.py
+14
View File
@@ -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
docs/getting-started/facility-automation.md
+4 -4
View File
@@ -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"
docs/reference/node-types.md
+16 -10
View File
@@ -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 |