A Rust port of ruby_zmq_framework and its Zig and Go siblings: Node-RED without the UI, built on Arc, Mutex, channels, and a plain trait for the node contract.

• Nodes are independent OS processes. Each one does one job, lives in one file, and knows nothing about any other node — not their ports, not their names, not their language.
• Wires are pub/sub topics carrying JSON, over ZeroMQ.
• The graph is data: flow.yml is the only artifact that knows the topology. flowctl reads it, computes the wiring, and runs everything.
• The contract is one page: PROTOCOL.md is everything a node in any language needs to join — and it's the same contract as the Ruby, Zig, and Go originals, so nodes from any of the four repos can sit in one flow.yml together without any of them knowing the others exist.

The node contract is a plain trait (Handler, one method, handle_message), checked by the compiler at the boot() call site — no reflection, no runtime contract check needed.

You need Rust (edition 2021+), a C compiler (for linking against libzmq — no zmq wrapper crate, see below), and the ZeroMQ library (libzmq3-dev on Debian/Ubuntu, brew install zeromq on macOS), then:

cargo build
target/debug/flowctl

That runs the demo graph from flow.yml: a simulated ECU blasting RPM data, a telemetry node that commands a throttle cut on over-rev, a web dashboard on http://localhost:4567, a state registry caching heartbeats and telemetry, and a dashboard consumer syncing the registry's snapshot. Output is streamed with a [node_name] prefix; Ctrl-C stops everything.

target/debug/flowctl --plan prints the computed wiring without running anything. target/debug/flowctl --graph prints the node topology as JSON.

A Rust node is a struct with one method, booted from the environment:

use rust_zmq_framework::{boot, sleep_forever, Bus, Handler};
use serde_json::{json, Value};
use std::sync::{Arc, Mutex};

// RpmSmoother publishes: engine_data_smooth. Subscribes: engine_data.
struct RpmSmoother {
    bus: Arc<Bus>,
    window: Mutex<Vec<i64>>,
}

impl Handler for RpmSmoother {
    fn handle_message(&self, topic: &str, payload: &Value) {
        if topic != "engine_data" {
            return;
        }
        let rpm = payload.get("rpm").and_then(|v| v.as_i64()).unwrap_or(0);

        let mut window = self.window.lock().unwrap();
        window.push(rpm);
        if window.len() > 5 {
            window.remove(0);
        }
        let avg = window.iter().sum::<i64>() / window.len() as i64;
        let _ = self.bus.publish("engine_data_smooth", &json!({ "rpm": avg }));
    }
}

fn main() {
    let (_handle, _smoother) =
        boot(|bus| RpmSmoother { bus, window: Mutex::new(Vec::new()) }).expect("boot failed");
    println!("online");
    sleep_forever();
}

Note what's absent: no ports, no peers, no subscribe calls. Wiring comes from environment variables (BUS_PORT, BUS_PEERS, BUS_SUBSCRIBES, NODE_NAME — see PROTOCOL.md), which flowctl computes from the node's entry in the manifest:

  rpm_smoother:
    cmd: target/debug/rpm_smoother
    subscribes: [engine_data]
    publishes: [engine_data_smooth]

(add src/bin/rpm_smoother.rs — Cargo auto-discovers binaries under src/bin/ — to get it built.)

Run standalone (no environment needed — it binds an ephemeral port) to poke at a node in isolation: target/debug/rpm_smoother.

Every node automatically heartbeats every 5 seconds. boot is generic (fn boot<N: Handler>(new_node: impl FnOnce(Arc<Bus>) -> N)); if a node's constructor needs more than the bus, close over the extra argument instead of adding a special case — see src/bin/can_bridge.rs. Because handle_message takes &self (handlers are shared via Arc so they can be reached from other threads too, e.g. an HTTP handler reading state a node updated), any state a node mutates from inside handle_message needs interior mutability — a Mutex field, as above, or an atomic for a single scalar.

The bus is just two-frame ZeroMQ pub/sub — [topic, json] — and the whole contract fits on one page: PROTOCOL.md, including a complete minimal Python node and the raw libzmq calls this framework's boot() makes under the hood. Follow it, add a cmd entry to flow.yml, and the language never matters again — including the original ruby_zmq_framework and its Zig, Go, Node, and C++ ports, which all speak the exact same wire format. flow_viewer can view and edit any of their flow.yml files.

Delivery is fire-and-forget (latest-value-wins), and a bad message or a panicking handler can never kill the dispatcher thread (std::panic::catch_unwind catches it, mirroring the other ports' per-handler error isolation). A handler that calls publish from within handle_message doesn't nest synchronously — it enqueues, and runs after the current message finishes and the dispatch loop cycles back around (the same breadth-first-not-depth-first tradeoff the Go port makes, for the same reason: it avoids needing a reentrant lock).

A correctness note relative to the Zig and Go ports: libzmq sockets are not safe to use concurrently from multiple threads. Any thread can call Bus::publish (the heartbeat thread, the main thread, a handler), and both the Zig and Go versions send on the shared PUB socket from whichever thread calls Publish with no synchronization around the actual zmq_send — technically undefined behavior under libzmq's own rules, even though it's unlikely to misbehave in a short demo run. Rust's Bus adds a Mutex around just the wire write (send_lock in src/bus.rs) to close that gap. Worth backporting if it matters for your use of those other ports.

Note: ZeroMQ is reached through a direct extern "C" binding (src/bus.rs) — no zmq wrapper crate, so this repo's only dependencies are serde/serde_json (JSON — not in std, unlike Go/Zig), libc (C type/syscall definitions), and rand (RNG — also not in std). The wire format is deliberately plain two-frame PUB/SUB, so swapping transports stays a contained change behind Bus's interface (publish/subscribe/close).

Uncomment the can_bridge node in flow.yml (set CAN_IFACE, e.g. vcan0) to relay real SocketCAN frames onto the bus as can_frame messages. Needs an actual or virtual CAN interface; fails fast if it doesn't exist.

cargo test

Unit tests live in #[cfg(test)] mod tests blocks alongside the code they cover (src/bus.rs, src/flow.rs, src/state_registry.rs, src/can_bridge.rs, src/framework.rs), mirroring the other ports' suites: bus dispatch, flow wiring/graph computation, StateRegistry's heartbeat/telemetry/snapshot behavior, and CAN frame parsing.