robot-bus 0.0.3

ZeroMQ message bus: broker routing and participant SDK

Robot Bus

轻量级、免环境配置的 ROS 2 风格通信库：基于 ZeroMQ，提供 topic / service / action，以及 Executor + Node + spin 回调模型。

不依赖 ROS 发行版、不需要 source setup.bash、不搭 workspace。一个 broker 进程 + SDK（Rust / Python）即可。

设计原则：API 会尽量贴近 ROS 2 的用法与命名（如 Node、SingleThreadedExecutor / MultiThreadedExecutor、add_node、create_publisher / create_subscription、spin），降低从 ROS 2 迁过来的心智负担；底层用 ZeroMQ 实现，不绑定某一 ROS 发行版。

预发布说明：当前仍处于预发布阶段。接下来 API 可能会有较多变更，运行稳定性也尚不完善，请谨慎用于生产环境。

更多 API 示例见 docs/。

模块	职责
broker::	路由进程（message / service / action）
顶层 API	Publisher / Subscriber / Client / Worker
runtime::Executor	底层 poll loop（一般用下面两个包装）
runtime::SingleThreadedExecutor / MultiThreadedExecutor	ROS 2 风格执行器；add_node + spin
runtime::Node / TopicPublisher / CallbackGroup	节点、publisher、callback group（互斥 / 可重入）
grpc::（默认 feature）	gRPC / gRPC-Web 网关（随 broker 一起启动）
proto/	ROS 风格 Protobuf：proto/<pkg>/{msg|srv|grpc}/v1/ → Rust/Python robot_bus.<pkg>…

架构

业务代码 (Rust / Python)
  └── robot-bus SDK
              │
              │ ZMQ (tcp / ipc / inproc)
              ▼
robot_bus_broker 进程

快速开始

1. 启动 broker

Rust：

cargo run --bin robot_bus_broker

Python（pip install robot-bus 后自带可执行入口）：

robot-bus-broker

或在代码里进程内启动：

import robot_bus

with robot_bus.RobotBusBroker.start() as broker:
    # ... 业务代码 ...
    pass
# 离开 with 自动 stop

# 或阻塞运行（等同于命令行，Ctrl+C 退出）
# robot_bus.run_broker()

Python

pip install robot-bus

本地开发（需 maturin）：

maturin develop --features extension-module,grpc

（grpc 为默认 feature；显式写出可避免 default-features = false 的构建漏掉网关。）

import robot_bus
from robot_bus.sensor_msgs.msg.v1 import Imu
from robot_bus.geometry_msgs.msg.v1 import Vector3

def on_imu(topic, payload):
    imu = Imu()
    imu.ParseFromString(payload)
    print(topic, imu.linear_acceleration)

node = robot_bus.Node("pilot")
executor = robot_bus.SingleThreadedExecutor()
executor.add_node(node)

imu_pub = node.create_publisher("/robot1/imu")
node.create_subscription("/robot1/imu", on_imu)
imu_pub.publish(
    Imu(linear_acceleration=Vector3(x=0.0, y=0.0, z=9.8)).SerializeToString(),
)
# executor.spin()  # 阻塞直到其它线程调用 executor.shutdown()

2. Rust（Executor + Node + spin）

在 Cargo.toml 中添加依赖：

robot-bus = { path = "../robot-bus" }
# 或 crates.io：robot-bus = "0.0.3"

语义接近 ROS 2：Node::new → executor.add_node → create_publisher(topic) → spin：

use std::sync::Arc;
use std::time::Duration;
use prost::Message;
use robot_bus::geometry_msgs::msg::v1::Vector3;
use robot_bus::sensor_msgs::msg::v1::Imu;
use robot_bus::{Node, SingleThreadedExecutor};

let mut node = Node::new("pilot");
let executor = SingleThreadedExecutor::new();
executor.add_node(&mut node)?;

let imu_pub = node.create_publisher("/robot1/imu")?;
node.create_subscription_typed::<Imu, _>(
    "/robot1/imu",
    |topic, imu| {
        println!("{topic}: {:?}", imu.linear_acceleration);
    },
    None,
)?;

let imu = Imu {
    linear_acceleration: Some(Vector3 { x: 0.0, y: 0.0, z: 9.8 }),
    ..Default::default()
};
imu_pub.publish(&imu.encode_to_vec())?;

node.create_timer(
    Duration::from_millis(100),
    Arc::new(|| {
        // 控制周期 / 心跳
    }),
    None,
)?;

let handle = executor.shutdown_handle()?;
std::thread::spawn(move || { /* ... */ handle.shutdown(); });
executor.spin()?;

• SingleThreadedExecutor：回调在 I/O / spin 线程串行（默认）
• MultiThreadedExecutor::new(n)：最多 n 个 worker；Reentrant group 可并行，MutuallyExclusive 组内串行
• Callback group：MutuallyExclusive / Reentrant（create_callback_group；默认互斥组）
• Service / action：create_service / create_client、create_action / create_action_client（与 topic 一样挂在 Node）
• Timer：create_timer（同样挂在 Node，由 spin 驱动）
• Raw bytes：create_subscription(topic, Arc::new(|topic, payload| { ... }), None)
• 底层 escape hatch：Executor（高级用法）

发送 / 接收水位（ZMQ HWM，不是完整 QoS）可在创建时或运行中设置：

use robot_bus::{Publisher, HighWaterMark};

let pub_ = Publisher::with_hwm(None, HighWaterMark::new(10, 10))?;
pub_.set_high_water_mark(HighWaterMark { snd: 10, rcv: 10 })?;

默认：message STREAM(2/2)、service RPC(4/4)、action ACTION(8/8)。Broker 侧用 --snd-hwm / --rcv-hwm。

二进制

二进制	说明
robot_bus_broker	一次启动三条总线 + gRPC / gRPC-Web 网关

gRPC / gRPC-Web 网关

随 robot_bus_broker / RobotBusBroker::start 一起启动；标准 gRPC 与 gRPC-Web 同端口（默认 0.0.0.0:15770）。

RPC	语义
MessageGateway.Subscribe	按 topic 前缀订阅，服务端流式返回二进制 payload
ServiceGateway.Call	一元：service_name + request bytes → response bytes
ActionGateway.Run	双向流：客户端发 GOAL / CANCEL，服务端推 ActionEvent（kind 区分 FEEDBACK / RESULT）

cargo run --bin robot_bus_broker
# 配置：cargo run --bin robot_bus_broker -- --help
# gRPC: http://0.0.0.0:15770

进程内：

use robot_bus::{GrpcBrokerConfig, RobotBusBroker, RobotBusConfig};

let broker = RobotBusBroker::start(RobotBusConfig {
    grpc: GrpcBrokerConfig {
        listen: "0.0.0.0:15770".parse()?,
        ..Default::default()
    },
    ..RobotBusConfig::default()
})?;
let grpc = format!("http://{}", broker.grpc_listen());

Proto（包名 robot_bus.grpc.v1，与 ROS *.msg.v1 / *.srv.v1 区分）：

• message_gateway.proto
• service_gateway.proto
• action_gateway.proto

测试

cargo test
PYTHONPATH=python python3 tests/python/test_msgs_roundtrip.py

Protobuf 消息

proto/ 按 ROS 包布局：proto/<pkg>/{msg|srv|grpc}/v1/*.proto。

语言	路径	说明
Rust	robot_bus::<pkg>::{msg|srv}::v1	build.rs + prost；挂在 crate 命名空间下
Python	robot_bus.<pkg>.{msg|srv}.v1	随 wheel 打包；scripts/generate_python_msgs.py 生成

• 传输层 body 仍是 opaque bytes；bus 不解析类型，业务侧自行 encode / decode（或用 create_subscription_typed）
• srv 是一对 *Request / *Response message，不是 gRPC
• grpc（robot_bus）是网关 RPC 契约，随 broker 启动（默认 feature grpc）
• 消息在 robot_bus 命名空间下，不占用 ROS 顶层 sensor_msgs 包名；编码是 protobuf，与 ROS CDR 不互通
• 改 proto 后跑：python3 scripts/generate_python_msgs.py（建议 protoc 28.x，与 CI 一致）

已覆盖：builtin_interfaces、std_msgs、std_srvs、geometry_msgs、sensor_msgs、nav_msgs、tf2_msgs、trajectory_msgs、diagnostic_msgs、unique_identifier_msgs、shape_msgs、visualization_msgs、control_msgs、nav2_msgs。