Merge remote-tracking branch 'origin/main'
This commit is contained in:
Melvin Strobl
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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/<flow>/<path>[/<secret>]`` 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/<flow>/<path>."
|
||||
)
|
||||
)
|
||||
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/<flow>/<url>/<secret>. 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)
|
||||
@@ -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": [],
|
||||
|
||||
@@ -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())
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user