The project is in a healthy, maintained state
Blackbox node processes wired together by pub/sub topics: the graph lives in a flow.yml manifest, nodes are configured entirely from the environment, and a one-page protocol lets any language join the bus. Includes a hardened ZeroMQ transport with strict runtime module contracts.
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
 Dependencies

Development

~> 5.20
~> 13.0

Runtime

~> 2.0
 Project Readme

ruby_zmq_framework

Think Node-RED without the UI: a flow-based, language-agnostic runtime for blackbox nodes wired together by topics.

  • 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. bin/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.

The design goal is LLM-friendliness: each node is a task you can hand to an agent with the protocol page and a one-line description ("write a Go node that decodes DBC frames") — it never needs the rest of the repo in context.

Quick start

You need the ZeroMQ library (libzmq3-dev on Debian/Ubuntu, brew install zeromq on macOS), then:

bundle install
bundle exec bin/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.

bundle exec bin/flowctl --plan prints the computed wiring without running anything. bundle exec bin/flowctl --graph prints the node topology as JSON — what the viewer below reads.

Viewing and editing the wiring

flow_viewer is a separate repo: a Node-RED-style graph viewer/editor for flow.yml — one box per node, colored wires between them for each topic, a dashed "unresolved" node wherever something subscribes to a topic nobody publishes, and an edit mode for dragging nodes, editing their publishes/subscribes/env, and wiring new connections, with an optional local server to save changes straight back to flow.yml. It's a separate repo because it only deals in the graph JSON shape and flow.yml text format every port here shares, so it isn't specific to this one:

git clone https://github.com/pgdaniel/flow_viewer
cd flow_viewer
npm install
npm run sync -- /path/to/ruby_zmq_framework/flow.yml
npm run dev

Writing a node

A Ruby node is a class with one method, booted from the environment:

require_relative '../lib/ruby_zmq_framework'
$stdout.sync = true

class RpmSmoother
  include RubyZmqFramework::FrameworkModule

  def initialize(bus)
    @bus = bus
    @window = []
  end

  def handle_message(topic, payload)
    @window = (@window << payload[:rpm]).last(5)
    broadcast(:engine_data_smooth, { rpm: @window.sum / @window.size })
  end
end

RubyZmqFramework.boot(RpmSmoother)
sleep

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: ruby nodes/rpm_smoother.rb
    subscribes: [engine_data]
    publishes: [engine_data_smooth]

Run standalone (no environment needed — it binds an ephemeral port) to poke at a node in isolation: bundle exec ruby nodes/rpm_smoother.rb.

Every node automatically heartbeats every 5 seconds. StrictContract raises at .new if a node forgets handle_message — loudly and immediately, which is exactly the feedback an iterating agent needs.

Nodes in other languages

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. Follow it, add a cmd entry to flow.yml, and the language never matters again. A Python companion library exists at python_zmq_framework, with the same boot()-from-environment contract, node_name, and clean shutdown as this gem — a Python node under its nodes/ and a Ruby node under this repo's nodes/ are interchangeable entries in one flow.yml.

The same is true of the other full ports of this framework — zig_zmq_framework, go_zmq_framework, rust_zmq_framework, node_zmq_framework, and cpp_zmq_framework — all speak the exact same wire format, so a node from any of them is an interchangeable entry in one flow.yml too. And flow_viewer can view and edit any of their flow.yml files, since it only deals in the topology they all share.

What's in the box

piece file job
ZeroMQBus lib/ruby_zmq_framework/zeromq_bus.rb hardened transport: poison-message-proof listener, per-handler error isolation, local dispatch, clean close
FrameworkModule lib/ruby_zmq_framework/framework.rb node mixin: contract enforcement, auto-heartbeat, broadcast, node_name, stop_heartbeat
Flow lib/ruby_zmq_framework/flow.rb parses flow.yml, computes each node's env wiring
flowctl bin/flowctl assigns ports, spawns nodes, prefixes output, tears down
StateRegistry lib/ruby_zmq_framework/state_registry.rb passive cluster-state cache; replays snapshots on request
CanBridge lib/ruby_zmq_framework/can_bridge.rb real SocketCAN frames → can_frame topic (classic CAN, via raw ioctls, no extra gem)
demo nodes nodes/*.rb one blackbox process per file

Delivery is fire-and-forget (latest-value-wins; slow consumers drop old messages), handlers on one bus never run concurrently, and a bad message or a raising handler can never kill a node's listener. See CHANGELOG.md for the full hardening history.

Note: ZeroMQ is reached through ffi-rzmq, which is stable but hasn't seen a release in years. The wire format is deliberately plain two-frame PUB/SUB, so swapping bindings — or the transport itself — stays a contained change behind the three-method bus interface (publish/subscribe/close).

CAN hardware

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 with the underlying Errno if it doesn't exist.

Tests

bundle exec rake test