ws:// and wss:// URLs dial out and hold a supervised connection; a path is served under /hooks/<flow>/ with an optional secret. Outputs only listen, inputs send. Frames are JSON keyed by port, with json_key and on_connect.
5.9 KiB
Flows, nodes and messages
A flow is a group of nodes that exchange messages. Nodes declare
what they consume (requires) and produce (provides); matching message names
connect them. No wires to draw.
How nodes connect
For a Function node, declare input and output ports in its panel, then write
process(...). Arguments use input port names; the returned dictionary uses
output port names:
# Node "read": provides temperature
def process():
return {"temperature": 18.5}
# Node "decide": consumes temperature, provides heat
def process(temperature, setpoint=21.0):
return {"heat": temperature < setpoint}
Bind both temperature ports to the same message and decide follows read.
Here setpoint uses a Python default; you can also add it as a node
setting.
- Several nodes can produce the same message. Consumers use the latest value.
- Publishing an input wakes its consumers once their required values are available. An upstream node that publishes nothing leaves downstream nodes waiting for a new publication.
- The editor arranges the graph automatically. Keep flows focused and share messages between them when needed.
Message names are namespaced
In flow house, temperature becomes house.temperature. A dotted name is
used as written, so another flow can consume it directly. Bind an input port
named temperature to house.temperature:
def process(temperature):
return {"shown": temperature}
The canvas shows cross-flow messages as labelled endpoints.
Node types and payloads
Function nodes run your Python code. Built-in nodes handle MQTT, HTTP, InfluxDB, schedules, filtering and more through forms in their panels. See Node types.
Every port declares a payload type (dtype). Values are checked against it,
and dashboard widgets offer compatible messages. Scalars, structured JSON and
media references are covered in Payload types.
Messages carry JSON. For binary data, store the bytes as an artifact and
publish its reference. image, audio and video are artifact references
restricted to their media family.
Streaming ports
Mark an output streaming when it publishes during an execution: progress,
training loss or media chunks. A Function node can yield values or call
fluksio.emit.
Downstream nodes receive each publication in order; message state holds the
latest value. A batch run records the full series. An output interval limits
live publication without thinning the run's recorded series.
Keeping node code predictable
Write Function nodes as functions of their declared inputs and settings.
Fluksio provides no hidden context or global flow store. To remember a value
between executions, use a message the node reads and writes; see
Keeping state in a flow.
Live and batch flows
Choose Mode in the flow settings:
live (default) |
batch |
|
|---|---|---|
| Runs | reacts continuously | once per requested run |
| Started by | subscriptions, schedules, webhooks or manual execution | Run in the editor, SDK or API |
| Finishes | stays active until stopped | when its work finishes |
| Keeps | latest message values | parameters, results, metrics and artifacts |
| Example | thermostat | experiment or build job |
Batch flows use the same editor and validation, but do not activate background subscriptions, schedules or webhooks. See Runs.
Flow inputs
Declare messages supplied by dashboard controls, the API or run parameters as the flow's Inputs. Give live inputs a starting value:
{"inputs": [{"spec": {"name": "setpoint", "dtype": "float"}, "initial": 21.0}]}
This seeds house.setpoint in flow house if it has no value yet. Later
publications replace it. Batch inputs are run parameters, and batch outputs
select the run's result. See Where a node's values come from.
Drafts and publishing
- Saving stores a draft; the engine keeps running the published version.
- Publishing promotes the draft and reloads the flow.
- Discarding removes the draft.
Shared node code is an exception: saving it updates every flow using it immediately. See The flow editor.
Flows are stored in a git repository: flow.json describes the graph and
nodes/*.py holds code. Saves are versioned, and stale versions return a
conflict instead of overwriting another editor's changes.
Python installations need Git on the host to record history. Without it, flows still save as files; Docker includes Git.
When a flow needs attention
The canvas reports issues on the affected nodes:
| Issue | Meaning |
|---|---|
unconnected_input |
no node or declared flow input supplies the message |
missing_initial_value |
a live input has no starting value |
cycle |
nodes depend on each other in a loop |
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 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 or advisory issues: other work continues, and node health clears when the connection recovers.
You can stop a live flow to remove its subscriptions and schedules, or pause it to hold pending work while values still arrive. Step releases one held item. Repeated background crashes can quarantine a flow; publishing a change gives it another chance.
Where to next
- Where a node's values come from — ports, settings and flow inputs
- Keeping state in a flow — remembering a value
- Runs — batch flows
- Writing node code — practical Python examples