From b18cb304926e7373df2f0da79e85f02ff7d82b08 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 14 Oct 2026 18:42:19 +0200 Subject: [PATCH] Add websocket node: client or served socket, listener or sender ws:// and wss:// URLs dial out and hold a supervised connection; a path is served under /hooks// with an optional secret. Outputs only listen, inputs send. Frames are JSON keyed by port, with json_key and on_connect. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011CT3mowxKeZywDCH4fS1gt --- backend/fluksio/flow/controller.py | 31 +- backend/fluksio/flow/nodes/__init__.py | 2 + backend/fluksio/flow/nodes/websocket.py | 419 ++++++++++++++++++++++ backend/tests/flow/test_node_types.py | 5 + backend/tests/flow/test_websocket_node.py | 136 +++++++ docs/concepts/flows.md | 2 +- docs/reference/node-types.md | 25 ++ frontend/src/components/Flow/FlowNode.tsx | 2 + 8 files changed, 612 insertions(+), 10 deletions(-) create mode 100644 backend/fluksio/flow/nodes/websocket.py create mode 100644 backend/tests/flow/test_websocket_node.py diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 393140c..3e8cdfc 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -26,7 +26,7 @@ from collections.abc import Callable from contextlib import suppress from dataclasses import dataclass, field from enum import Enum -from typing import Any, cast +from typing import Any, TypeGuard, cast from fastapi import FastAPI from fastapi.concurrency import run_in_threadpool @@ -54,6 +54,7 @@ from fluksio.flow.nodes import ( RbeNode, SwitchNode, TriggerNode, + WebSocketNode, WrenNode, ) from fluksio.flow.pipeline import ( @@ -93,6 +94,14 @@ logger = logging.getLogger(__name__) 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 ( + isinstance(node, WebSocketNode) and node.serves + ) + + # How long a node gets to close what it opened before the rebuild moves on. # A node's `stop` talks to whatever it connected to, and a broker that has gone # away can leave it waiting for an acknowledgement that never arrives — which @@ -339,6 +348,12 @@ NODE_TYPES: dict[str, NodeType] = { cls=HttpNode, params_schema=_schema_of(HttpNode), ), + "websocket": NodeType( + title="WebSocket", + description="Listen on a socket, or send to it — as a client or a server.", + cls=WebSocketNode, + params_schema=_schema_of(WebSocketNode), + ), "influxdb": NodeType( title="InfluxDB", description="Write measurements to a bucket, or read them back.", @@ -1229,8 +1244,8 @@ class FlowController: node.assign_flow(flow, node_def.id) node._on_health = self._health_changed - if isinstance(node, HttpNode) and node.mode == HttpNode.Mode.TRIGGER: - # Webhooks live under their flow, away from the JSON API. + if _is_hook(node): + # Hooks live under their flow, away from the JSON API. node.url = f"{HOOK_PREFIX}/{flow}/{node.url.lstrip('/')}" entry.node = node @@ -1876,23 +1891,21 @@ def _collect_issues( for entry in loaded.values() if entry.status is NodeStatus.ERROR ] - # Webhooks are mounted without authentication, so a hook without a secret is + # Hooks are mounted without authentication, so a hook without a secret is # open to anyone who guesses its URL. Flows written before the parameter # existed keep running, but say so. issues += [ ValidationIssue( code="unauthenticated_hook", message=( - f"Webhook '{entry.node.local_id}' has no secret — " - "anyone who knows its URL can trigger it." + f"Hook '{entry.node.local_id}' has no secret — " + "anyone who knows its URL can reach it." ), flow=entry.flow, node=entry.id, ) for entry in loaded.values() - if isinstance(entry.node, HttpNode) - and entry.node.mode == HttpNode.Mode.TRIGGER - and not entry.node.secret + if _is_hook(entry.node) and not entry.node.secret ] # A node whose body was never stored runs the new-node template, which # returns nothing — quietly, and looking healthy the whole time. diff --git a/backend/fluksio/flow/nodes/__init__.py b/backend/fluksio/flow/nodes/__init__.py index 808cb4f..4f91e03 100644 --- a/backend/fluksio/flow/nodes/__init__.py +++ b/backend/fluksio/flow/nodes/__init__.py @@ -16,6 +16,7 @@ from fluksio.flow.nodes.mlp import MLPNode from fluksio.flow.nodes.mqtt import MqttNode from fluksio.flow.nodes.ntfy import NtfyNode from fluksio.flow.nodes.trigger import TriggerNode +from fluksio.flow.nodes.websocket import WebSocketNode from fluksio.flow.nodes.wren import WrenNode __all__ = [ @@ -35,5 +36,6 @@ __all__ = [ "RbeNode", "SwitchNode", "TriggerNode", + "WebSocketNode", "WrenNode", ] diff --git a/backend/fluksio/flow/nodes/websocket.py b/backend/fluksio/flow/nodes/websocket.py new file mode 100644 index 0000000..15732f6 --- /dev/null +++ b/backend/fluksio/flow/nodes/websocket.py @@ -0,0 +1,419 @@ +"""WebSocket nodes: a socket the engine dials or serves, listening or sending.""" + +from __future__ import annotations + +import asyncio +import contextlib +import hmac +import json +import logging +from collections.abc import Awaitable, Callable, Iterable +from enum import Enum +from typing import TYPE_CHECKING, Any +from urllib.parse import urlsplit + +from pydantic import BaseModel, ConfigDict, Field + +from fluksio.flow.messages import MessageSpec +from fluksio.flow.nodes.base import Node +from fluksio.flow.nodes.http import HttpNode, _first_catch_all + +if TYPE_CHECKING: + from fastapi import FastAPI + from starlette.websockets import WebSocket + +logger = logging.getLogger(__name__) + + +class WebSocketNode(Node): + """ + A long-lived WebSocket, as a client or a server. + + The URL picks the role: ``ws://`` or ``wss://`` dials out and holds the + connection, a path such as ``/ticker`` is served at + ``/hooks//[/]`` while the flow runs. The ports pick the + direction, as on the mqtt node: outputs only make a **listener**, inputs a + **sender**. + + - client listener: frames the server sends become outputs + - client sender: inputs go to the server, one frame per message + - server listener: frames any connected client sends become outputs + - server sender: inputs go to every connected client + + Frames are text. An incoming frame is parsed as JSON; an object feeds the + ports whose names (or ``json_key``) it carries, and a node with one output + takes anything else whole. Outgoing, a single input is sent as it stands — + a string raw, anything else as JSON — and several go as one object keyed + by port. + """ + + class Mode(Enum): + """Which way the frames go.""" + + LISTENER = "listener" + SENDER = "sender" + + # A frame sent twice is two commands to whatever reads it. + idempotent = False + + __slots__ = ( + "url", + "mode", + "serves", + "headers", + "on_connect", + "secret", + "timeout", + "send_queue_size", + "json_keys", + "_clients", + "_route_registered", + "_task", + "_queue", + "_loop", + ) + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + url: str = Field( + description=( + "ws:// or wss:// to connect to a server, or a path such as " + "/ticker to serve one at /hooks//." + ) + ) + headers: dict[str, str] = Field( + default_factory=dict, + description=( + "Handshake headers, client only. A value may be a secret reference." + ), + ) + on_connect: list[Any] = Field( + default_factory=list, + description=( + "Frames sent after every (re)connect, client only — the " + "subscribe or auth message many APIs expect. A string goes " + "raw, anything else as JSON; an item may be a secret reference." + ), + ) + secret: str = Field( + default="", + description=( + "Shared secret for a served socket, appended to its URL: " + "/hooks///. Empty leaves it open to anyone." + ), + json_schema_extra={"x-secret": True}, + ) + timeout: float = Field( + default=10.0, + gt=0, + description=( + "Seconds allowed for the handshake, and for a send to one " + "client of a served socket." + ), + ) + send_queue_size: int = Field( + default=256, + gt=0, + description=( + "How many messages may wait to be sent. Past this the oldest " + "is dropped and the node reports degraded." + ), + ) + # Same as the mqtt node: a device that wraps its reading in an object. + json_key: str | dict[str, str] = "" + + @classmethod + def target_key(cls, params: dict[str, Any]) -> str | None: + # A URL, credentials stripped, exactly as for the http node. + return HttpNode.target_key(params) + + def __init__( + self, + requires: MessageSpec | Iterable[MessageSpec] = (), + provides: MessageSpec | Iterable[MessageSpec] = (), + params: dict[str, Any] | None = None, + name: str | None = None, + ): + cfg = self.Params.model_validate(params or {}) + + requires = Node._normalize_ports(requires) + provides = Node._normalize_ports(provides) + if not requires and not provides: + raise ValueError( + "A WebSocket node needs either inputs (to send) or outputs (to listen)" + ) + # ponytail: one direction per node, as mqtt and http. Inputs and + # outputs over one socket is the NOTEPAD follow-up. + self.mode = ( + WebSocketNode.Mode.SENDER if requires else WebSocketNode.Mode.LISTENER + ) + + self.serves = cfg.url.startswith("/") + if not self.serves and urlsplit(cfg.url).scheme not in ("ws", "wss"): + raise ValueError( + "A WebSocket url is ws://…, wss://… or a path to serve, " + f"not '{cfg.url}'" + ) + + ports = requires or provides + if isinstance(cfg.json_key, dict): + self.json_keys: dict[str, str] = dict(cfg.json_key) + else: + self.json_keys = ( + {spec.port: cfg.json_key for spec in ports} if cfg.json_key else {} + ) + + self.url = cfg.url + self.headers = cfg.headers + self.on_connect = [_frame(item) for item in cfg.on_connect] + self.secret = cfg.secret + self.timeout = cfg.timeout + self.send_queue_size = cfg.send_queue_size + + self._clients: set[WebSocket] = set() + self._route_registered = False + self._task: asyncio.Task[None] | None = None + self._queue: asyncio.Queue[dict[str, Any]] | None = None + self._loop: asyncio.AbstractEventLoop | None = None + + if name is None: + safe_url = ( + self.url.split("://", 1)[-1].replace("/", "_").replace(":", "") + ).strip("_") + name = f"websocket_{self.mode.value}_{safe_url}" + + super().__init__( + f=( + self._sender_handler + if self.mode is WebSocketNode.Mode.SENDER + else self._noop_listener + ), + requires=requires, + provides=provides, + params=params, + name=name, + ) + + @staticmethod + def _noop_listener(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + """A listener injects what arrives; it is never called with inputs.""" + return None + + # -- frames --------------------------------------------------------------- + + def _encode(self, data: dict[str, Any]) -> str: + """One message as one frame.""" + if len(data) == 1: + return _frame(next(iter(data.values()))) + return json.dumps(data, default=str) + + def _decode(self, text: str) -> dict[str, Any]: + """One frame as outputs, keyed by port. Empty when nothing matched.""" + try: + parsed = json.loads(text) + except json.JSONDecodeError: + parsed = text + outputs: dict[str, Any] = {} + if isinstance(parsed, dict): + for spec in self.output_ports: + key = self.json_keys.get(spec.port, spec.port) + if key in parsed: + outputs[spec.port] = spec.coerce(parsed[key]) + if not outputs and len(self.output_ports) == 1: + spec = self.output_ports[0] + outputs[spec.port] = spec.coerce(parsed) + return outputs + + async def _receive(self, text: str) -> None: + """Inject one frame. A bad one is logged, never fatal to the socket.""" + try: + outputs = self._decode(text) + if outputs: + await asyncio.to_thread(self.inject, outputs) + else: + logger.debug("[%s] Frame matched no port: %s", self.name, text) + except Exception as exc: + logger.error("[%s] Error processing frame: %s", self.name, exc) + + # -- sending -------------------------------------------------------------- + + def _sender_handler( + self, params: dict[str, Any], **kwargs: Any + ) -> dict[str, Any] | None: + """Hand the inputs to the node's socket task, which owns the send.""" + loop, queue = self._loop, self._queue + if loop is not None and queue is not None: + loop.call_soon_threadsafe(self._enqueue, queue, dict(kwargs)) + return None + + # Not started: a draft preview or a test. A client connects once and + # sends; a served socket has nobody to send to. + if not self.serves: + from websockets.sync.client import connect + + with connect( + self.url, + additional_headers=self.headers, + open_timeout=self.timeout, + ) as ws: + for frame in self.on_connect: + ws.send(frame) + ws.send(self._encode(dict(kwargs))) + return None + + def _enqueue( + self, queue: asyncio.Queue[dict[str, Any]], data: dict[str, Any] + ) -> None: + """Queue a message, dropping the oldest when the socket cannot keep up.""" + if queue.full(): + with contextlib.suppress(asyncio.QueueEmpty): + queue.get_nowait() + logger.warning("Send queue full for node '%s', dropped", self.name) + self.report_health("degraded", "send queue full") + queue.put_nowait(data) + + async def _send_queued(self, send: Callable[[str], Awaitable[None]]) -> None: + """Send whatever is queued, in order, for as long as the socket lives.""" + queue = self._queue + if queue is None: + return + while True: + await send(self._encode(await queue.get())) + + async def _broadcast(self, text: str) -> None: + """One frame to every connected client. + + ponytail: sequential, so a stalled client holds the others up for + `timeout`. Per-client queues if a served socket ever has many readers. + """ + for ws in list(self._clients): + try: + await asyncio.wait_for(ws.send_text(text), self.timeout) + except Exception: + self._clients.discard(ws) + with contextlib.suppress(Exception): + await asyncio.wait_for(ws.close(code=1011), self.timeout) + + # -- client --------------------------------------------------------------- + + async def _client_loop(self) -> None: + """Hold one connection. A drop raises, and the supervisor reconnects.""" + from websockets.asyncio.client import connect + + try: + async with connect( + self.url, + additional_headers=self.headers, + open_timeout=self.timeout, + ) as ws: + for frame in self.on_connect: + await ws.send(frame) + self.report_health("ok") + + if self.mode is WebSocketNode.Mode.LISTENER: + async for received in ws: + if isinstance(received, str): + await self._receive(received) + else: + sender = asyncio.create_task(self._send_queued(ws.send)) + try: + # Replies are not a sender's to publish, but they are + # read: an unread buffer stops the pings being answered. + async for _ in ws: + pass + finally: + await self._cancel_task(sender) + except asyncio.CancelledError: + raise + except Exception as exc: + self.report_health("down", str(exc)) + raise + self.report_health("down", "connection closed") + raise ConnectionError(f"WebSocket for '{self.name}' closed") + + # -- server --------------------------------------------------------------- + + async def _serve(self, websocket: WebSocket) -> None: + """One client of a served socket, for as long as it stays.""" + if self.secret and not hmac.compare_digest( + str(websocket.path_params.get("secret", "")).encode(), + self.secret.encode(), + ): + # Closed before accepting: the same 403 as a path with no route. + await websocket.close(code=1008) + return + await websocket.accept() + self._clients.add(websocket) + try: + while True: + message = await websocket.receive() + if message["type"] == "websocket.disconnect": + break + text = message.get("text") + # A sender's clients are read only to notice them leave. + if text is not None and self.mode is WebSocketNode.Mode.LISTENER: + await self._receive(text) + finally: + self._clients.discard(websocket) + + def register_route(self, app: FastAPI) -> None: + """Mount the served socket, ahead of any catch-all, as webhooks are.""" + if self._route_registered: + return + from starlette.routing import WebSocketRoute + + # The secret is a path parameter, so the route table never holds it. + path = f"{self.url}/{{secret}}" if self.secret else self.url + app.routes.insert( + _first_catch_all(app), WebSocketRoute(path, self._serve, name=self.id) + ) + self._route_registered = True + logger.info("Registered websocket route for node '%s': %s", self.name, path) + + def unregister_route(self, app: FastAPI) -> None: + if not self._route_registered: + return + app.routes[:] = [ + route for route in app.routes if getattr(route, "name", None) != self.id + ] + self._route_registered = False + + # -- lifecycle ------------------------------------------------------------ + + async def start(self, app: FastAPI | None = None) -> None: + if self.serves and app is not None: + self.register_route(app) + if self.mode is WebSocketNode.Mode.SENDER and self._queue is None: + # Outlives a reconnect, so what was queued meanwhile still goes. + self._queue = asyncio.Queue(maxsize=self.send_queue_size) + self._loop = asyncio.get_running_loop() + # Spawned on every start: enabling a flow cancels its supervised loops + # and starts the nodes again, and the supervisor ignores a duplicate. + if self._task is not None and not self._task.done(): + return + if not self.serves: + self._task = self._run_supervised("websocket", self._client_loop) + elif self.mode is WebSocketNode.Mode.SENDER: + self._task = self._run_supervised( + "websocket-out", lambda: self._send_queued(self._broadcast) + ) + + async def stop(self, app: FastAPI | None = None) -> None: + if self.serves and app is not None: + self.unregister_route(app) + for ws in list(self._clients): + with contextlib.suppress(Exception): + await ws.close(code=1001) + self._clients.clear() + if self._task is not None: + await self._cancel_task(self._task) + self._task = None + self._queue = None + self._loop = None + + +def _frame(value: Any) -> str: + """A value as frame text: a string as it stands, anything else as JSON.""" + # default=str: a value JSON cannot hold must not kill the send loop. + return value if isinstance(value, str) else json.dumps(value, default=str) diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index 27b7421..00b9d7b 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -17,6 +17,11 @@ FIXTURES: dict[str, dict] = { "requires": [], "provides": [MessageSpec(name="temp", dtype=DType.FLOAT)], }, + "websocket": { + "params": {"url": "wss://feed.local/live"}, + "requires": [], + "provides": [MessageSpec(name="temp", dtype=DType.FLOAT)], + }, "http": { "params": {"url": "/hook", "method": "POST"}, "requires": [], diff --git a/backend/tests/flow/test_websocket_node.py b/backend/tests/flow/test_websocket_node.py new file mode 100644 index 0000000..8fb7df2 --- /dev/null +++ b/backend/tests/flow/test_websocket_node.py @@ -0,0 +1,136 @@ +"""The websocket node, both roles: served under /hooks, or dialled out.""" + +import asyncio +import json + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from starlette.websockets import WebSocketDisconnect +from websockets.asyncio.server import ServerConnection, serve + +from fluksio.flow.controller import HOOK_PREFIX +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.nodes import WebSocketNode +from fluksio.flow.pipeline import Pipeline + +SECRET = "s0cket-key" +HOOK = f"{HOOK_PREFIX}/house/ticker" + + +def ws_node(url: str, *, sends: bool = False, **params: object) -> WebSocketNode: + port = [MessageSpec(name="temp", dtype=DType.FLOAT)] + node = WebSocketNode( + requires=port if sends else (), + provides=() if sends else port, + params={"url": url, **params}, + name="ticker", + ) + node.assign_flow("house", "ticker") + if node.serves: + node.url = HOOK # what the controller does + return node + + +def test_frames_map_to_ports(): + node = ws_node("ws://x", json_key="value") + assert node._decode('{"value": "21.5"}') == {"temp": 21.5} + # One output takes a frame that names no port whole. + assert node._decode("19") == {"temp": 19.0} + + sender = ws_node("ws://x", sends=True) + assert sender._encode({"temp": 21.5}) == "21.5" + assert sender._encode({"cmd": {"op": "on"}}) == '{"op": "on"}' + assert json.loads(sender._encode({"a": 1, "b": "x"})) == {"a": 1, "b": "x"} + + with pytest.raises(ValueError): + ws_node("http://x") + + +def test_a_served_socket_injects_frames_and_guards_its_secret(): + node = ws_node("/ticker", secret=SECRET) + pipeline = Pipeline(nodes=[node]) + app = FastAPI() + app.mount("/", FastAPI()) # the MCP catch-all the route must sit ahead of + node.register_route(app) + client = TestClient(app) + + with pytest.raises(WebSocketDisconnect): + with client.websocket_connect(f"{HOOK}/wrong") as ws: + ws.receive_text() + + with client.websocket_connect(f"{HOOK}/{SECRET}") as ws: + ws.send_text('{"temp": 21.5}') + assert pipeline.values()["house.temp"]["value"] == 21.5 + + +def test_a_served_sender_reaches_every_connected_client(): + node = ws_node("/ticker", sends=True) + app = FastAPI() + + with TestClient(app) as client: + client.portal.call(node.start, app) + with ( + client.websocket_connect(HOOK) as first, + client.websocket_connect(HOOK) as second, + ): + # Both are registered once either has been accepted and spoken to. + client.portal.call(asyncio.sleep, 0.05) + node._sender_handler({}, temp=21.5) + assert first.receive_text() == "21.5" + assert second.receive_text() == "21.5" + client.portal.call(node.stop, app) + + +def test_a_client_listens_and_sends_over_a_held_connection(): + received: list[str] = [] + + async def server(conn: ServerConnection) -> None: + if conn.request is not None and conn.request.path == "/feed": + await conn.send('{"temp": 19.5}') + await conn.wait_closed() + else: + async for frame in conn: + received.append(str(frame)) + + async def scenario() -> None: + async with serve(server, "127.0.0.1", 0) as srv: + port = srv.sockets[0].getsockname()[1] + listener = ws_node(f"ws://127.0.0.1:{port}/feed") + pipeline = Pipeline(nodes=[listener]) + sender = ws_node( + f"ws://127.0.0.1:{port}/cmd", + sends=True, + on_connect=[{"op": "subscribe"}], + ) + await listener.start() + await sender.start() + await asyncio.to_thread(sender._sender_handler, {}, temp=21.5) + for _ in range(100): + if len(received) == 2 and pipeline.values(): + break + await asyncio.sleep(0.02) + await listener.stop() + await sender.stop() + + assert received == ['{"op": "subscribe"}', "21.5"] + assert pipeline.values()["house.temp"]["value"] == 19.5 + + asyncio.run(scenario()) + + +def test_start_respawns_a_loop_cancelled_underneath_it(): + """Enabling a flow cancels its supervised loops and starts the nodes again, + without a stop in between.""" + node = ws_node("/ticker", sends=True) + + async def scenario() -> None: + await node.start() + first = node._task + assert first is not None + await node._cancel_task(first) + await node.start() + assert node._task is not first and not node._task.done() + await node.stop() + + asyncio.run(scenario()) diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index 6056cc8..70353cd 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -131,7 +131,7 @@ The canvas reports issues on the affected nodes: | `self_loop_needs_initial` | a node reads and writes a message with no starting value | | `missing_source` | a Function node has no saved code | | `node_error` | code or configuration could not load | -| `unauthenticated_hook` | a webhook has no shared secret; advisory | +| `unauthenticated_hook` | a webhook or served websocket has no shared secret; advisory | | `node_unhealthy` | a loaded node reports a runtime problem, such as a lost connection | Build issues prevent the affected flow from running. The last two are runtime diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index 6d70d50..7366d04 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -75,6 +75,31 @@ Outputs only makes it a **webhook**: the engine mounts a route at The canvas flags this as an advisory issue rather than stopping the flow, because a webhook on a private network is a legitimate thing to want. +### WebSocket + +**`websocket`** — listen on a socket, or send to it, as a client or a server. + +The URL picks the role: `ws://…` or `wss://…` connects to a server and +reconnects when it drops; a path such as `/ticker` is served at +`/hooks/{flow}/{url}/{secret}` while the flow runs. The ports pick the +direction: outputs only make a **listener**, inputs a **sender**. A served +sender sends to every connected client. + +| Setting | Default | Notes | +|---|---|---| +| `url` | — | `ws://` or `wss://` URL, or the path to serve | +| `headers` | `{}` | handshake headers, client only; a value may be a secret reference | +| `on_connect` | `[]` | frames sent after every (re)connect, client only — a subscribe or auth message; an item may be a secret reference | +| `secret` | — | shared secret appended to a served socket's URL; takes a secret reference | +| `timeout` | `10.0` | seconds for the handshake, and for a send to one client of a served socket | +| `send_queue_size` | `256` | messages that may wait to be sent; past this the oldest is dropped and the node reports degraded | +| `json_key` | — | as on `mqtt` | + +Frames are text. An incoming frame is parsed as JSON: an object feeds the +ports it names, and a node with a single output takes any other frame whole. +A single input is sent as it stands — a string raw, anything else as JSON — +and several go as one object keyed by port. + ### InfluxDB **`influxdb`** — write measurements to a bucket, or read them back. diff --git a/frontend/src/components/Flow/FlowNode.tsx b/frontend/src/components/Flow/FlowNode.tsx index f7511d4..d40dbaf 100644 --- a/frontend/src/components/Flow/FlowNode.tsx +++ b/frontend/src/components/Flow/FlowNode.tsx @@ -4,6 +4,7 @@ import { Bird, Braces, Bug, + Cable, Clock, Code2, Database, @@ -45,6 +46,7 @@ const NODE_ICONS = { python: Code2, mqtt: Radio, http: Globe, + websocket: Cable, influxdb: Database, delay: Clock, mlp: Braces,