跳到主要内容

RFrame 开发指南

文档总入口：《RFrame 使用入口》

1. 文档定位​

本文说明 RFrame 业务代码的开发方式，包括：

  • AppCore 生命周期；
  • ChannelConfigLoader；
  • ChannelManager；
  • typed HybridChannel；
  • typed HybridRpcChannel；
  • HybridChannel / HybridRpcChannel 用法清单（HMI 桥接、相机与 RobbyMotion 接口落地，见 §7）；
  • 生命周期托管和错误处理。

SDK 和示例如何编译见《RFrame 编译指南》；JSON 字段和部署方式见《RFrame 部署指南》。

可运行示例：安装 deb 包 robby-r-frame-example 后位于 /opt/robby/RFrameExample/，安装树如下：

/opt/robby/RFrameExample/
├── README.md
├── config/                     # 演示场景 channel 配置（interop / shm 调优等）
│   └── *.json
├── src/                        # C++ 示例源码
│   ├── channel/
│   │   ├── quickstart/         # 最小上手：channel_manager / hybrid_channel
│   │   ├── config/             # JSON 加载与上线前检查：channel_config_json_example
│   │   ├── typed/              # typed topic / typed RPC，主推写法
│   │   ├── advanced/           # 回调组、重启、双工、文件通道、零拷贝
│   │   ├── interop/            # hybrid 跨后端互通
│   │   └── rpc/                # RPC SHM / IPC
│   └── log/                    # 日志配置与格式（§2.1）
└── python/                     # Python 示例（topic / RPC / 跨语言互通）+ README.md

如何从源码编译出可执行文件见《RFrame 编译指南》。

1.1. 用户开发目录与持久化​

当前提供的开发容器中，/mnt/ssd 为持久化路径。建议用户自行创建 /mnt/ssd/workspace，将业务源码、配置文件、编译产物以及需要保留的日志和数据放在该目录下。

mkdir -p /mnt/ssd/workspace
cd /mnt/ssd/workspace

注意：容器重启会清理 /mnt/ssd 以外路径中的用户文件和修改。 开发时请先将需要修改的示例源码复制到 /mnt/ssd/workspace，不要将重要文件仅保存在容器内的其他目录。

2. 应用初始化​

AppCore::init() 应是第一项 RFrame 操作：

#include "r/frame/app/AppCore.h"
#include "r/frame/channel/ChannelConfigLoader.h"
#include "r/frame/channel/ChannelManager.h"

using namespace r::frame;

int main()
{
    auto app = AppCore::instance();
    if (app->init() != 0) {
        return 1;
    }

    app->addPlugin(ChannelManager::instance());

    // 加载配置、创建 Channel、执行业务。

    app->shutdown();
    return 0;
}

AppCore::init() 负责加载全局日志配置和完成框架初始化。Channel 配置由业务代码独立加载。

2.1. 业务日志（C++）​

业务打日志用 ROBBY_USER_* 宏（user 流；ROBBY_SYS_* 是框架系统流，业务一般不用），fmt 风格 {} 占位符：

#include "r/log/Log.h"

ROBBY_USER_INFO("channel {} opened, topic={}", name, topic);
ROBBY_USER_WARN("retry {} failed", n);
ROBBY_USER_ERROR("call {} timeout", method);

级别共 trace / debug / info / warn / error / critical 六档。注意：trace/debug 宏受编译期 SPDLOG_ACTIVE_LEVEL 门控，默认只编入 info 及以上，运行时调级别也救不回来；需要 debug 输出时编译加 -DSPDLOG_ACTIVE_LEVEL=SPDLOG_LEVEL_DEBUG。

不写任何日志代码也有日志：AppCore::init() 会用默认配置初始化日志引擎（默认写 ./log 目录；app->init(log_dir) 可指定目录），级别由全局 rframe.json 的 log.level 或 RFRAME_LOG_LEVEL 控制（见《RFrame 部署指南》§5）。

要自定义目录、分流分级和滚动，用 LogConfig + LogCore::init()，必须早于 AppCore::init()——否则 AppCore 先用默认配置把引擎初始化掉，后到的 init(cfg) 会被去重拦截而不生效：

r::log::LogConfig cfg;
cfg.app_name = "my_app";                        // 日志落在 <log_dir>/<app_name>/
cfg.log_dir  = "/var/log/my_app";
cfg.mode     = r::log::LogMode::Async;          // 高吞吐用异步
cfg.user_console_level = r::log::Level::info;   // user 流控制台级别
cfg.user_file_level    = r::log::Level::debug;  // user 流文件级别（可与控制台不同）
cfg.max_file_size      = 10ull * 1024 * 1024;   // 单文件滚动大小
cfg.total_limit_bytes  = 200ull * 1024 * 1024;  // 总量上限
r::log::LogCore::init(cfg);

app->init();   // 日志引擎已就绪，AppCore 只接管生命周期

运行时还可调：LogCore::setLoggerLevel(ROBBY_LOG_USER, ...) 调级别、setLogFormat() 换格式、flushAllLogs() 强制落盘、currentConfig() 回读生效配置；同目录跑多个进程时先用 setLoggerNames() 给各进程起不同 logger 名，避免日志文件互踩。Python 侧用法见 §8.1。

完整可运行参考：/opt/robby/RFrameExample/src/log/log_format_example.cpp（配置全貌 + 运行时调参）。

3. 加载配置​

3.1. 从独立文件加载​

auto configs = ChannelConfigLoader::loadFromFile("config/channels.json");
auto one = ChannelConfigLoader::loadByName(
    "config/channels.json", "head_color_reader");
if (configs.empty()) {
    app->shutdown();
    return 2;
}

3.2. 从字符串加载​

auto configs = ChannelConfigLoader::loadFromString(json_string);
auto one = ChannelConfigLoader::loadSingleChannel(
    "head_color_reader", json_string);

字符串入口适合配置中心下发和测试。可复用配置仍建议保留 {"channels": {...}} 外层。

加载结果为空时，应检查配置路径、JSON 语法、channels 外层，以及目标机安装的 RFrame runtime 是否完整。JSON loader 的内部实现依赖不作为业务配置接口暴露。

3.3. 从全局配置文件加载​

auto configs = ChannelConfigLoader::loadFromGlobalConfig();

不显式传路径，从 GlobalConfig::findConfigFile() 定位的全局配置文件里读 channels 段。定位顺序：RFRAME_CONFIG_PATH 环境变量 → ./rframe.json → ~/.config/rframe/rframe.json。

适合单进程只用一份全局配置的部署形态；需要按进程分别加载多份适配文件时（如 §7.4 按清单组装的 channels.json），仍用 §3.1 显式指定路径，互不干扰。

3.4. 代码内创建配置（不走 JSON）​

配置也可以直接在代码里构造，经同一个 ChannelManager::createChannel() 创建，生命周期同样由 ChannelManager 托管。适用场景：快速原型、单元测试、参数硬编码的小工具。正式交付仍推荐 JSON 文件：清单即事实源、部署期可调整、可用 channel_config_json_example 做上线前检查（《RFrame 部署指南》§8）。

推荐用工厂方法创建，子配置按合理默认值自动生成：

#include "r/frame/channel/HybridMsgChannel.h"
#include "r/frame/channel/HybridRpcChannel.h"

using namespace r::frame;

// ---- HybridChannel（topic）----
// SHM-only，64KB 共享内存（默认）
auto cfg = HybridChannelConfig::createShmOnly(
    "/demo/hello", FileHandlerType::READER);
auto ch = ChannelManager::instance()->createChannel("demo_sub", cfg);

// 双通路（SHM + ROS 2，发现驱动自动路由）
auto dual = HybridChannelConfig::create(
    "/demo/hello", FileHandlerType::WRITER,
    /*enable_shm=*/true, /*enable_ros2=*/true, /*auto_route=*/true);

// 其他工厂：createRos2Only / createZenoh / createZenohOnly

// ---- HybridRpcChannel（service）----
// SHM-only，默认超时 3000ms
auto rpc_cfg = HybridRpcChannelConfig::createShmOnly("/demo/rpc", 3000);
auto rpc = ChannelManager::instance()->createChannel("demo_rpc", rpc_cfg);
rpc->rpc()->registerHandler("echo", /* handler */);

// 双通路
auto rpc_dual = HybridRpcChannelConfig::create(
    "/demo/rpc", /*enable_shm=*/true, /*enable_ros2=*/true,
    /*auto_route=*/true, /*default_timeout_ms=*/3000);

// 其他工厂：createRos2Only

需要细调时改子配置字段（工厂返回后再覆盖）：

// topic：成员直接可写
cfg->shm_config->memory_size = 1 << 20;          // SHM 容量
cfg->ros2_config->type_name = "std_msgs/msg/String";  // 启用 ROS 2 时必填类型
cfg->ros2_config->node_name = "my_node";

// RPC：经 accessor 调整
rpc_cfg->shmBufferSize() = 1 << 20;   // RPC 大报文时上调（默认 64KB，见 §7.10）
rpc_cfg->defaultTimeoutMs() = 8000;

JSON 键与代码字段一一对应（如 hybrid.enable_remote ↔ enable_ros2、hybrid_rpc.shm_buffer_size ↔ shmBufferSize()），两种写法可混合：配置文件为主，代码只覆盖少量字段。

完整可编译参考：/opt/robby/RFrameExample/src/channel/typed/typed_channel_example.cpp、typed_rpc_example.cpp（按 --backend shm|ros2|dual 选择工厂方法），手工逐字段填充子配置的写法见 src/channel/quickstart/hybrid_channel_example.cpp。

4. 通过 ChannelManager 创建 Channel​

const std::string channel_name = "head_color_reader";
auto it = configs.find(channel_name);
if (it == configs.end() || !it->second || !it->second->isValid()) {
    app->shutdown();
    return 3;
}

auto manager = ChannelManager::instance();
auto channel = manager->createChannel(
    channel_name, it->second, true);
if (!channel) {
    app->shutdown();
    return 4;
}

第三个参数 true 表示创建后立即打开。实例名称必须与 JSON 中 channels 下的键一致。

批量创建时不要静默忽略失败：

std::unordered_map<std::string, std::shared_ptr<MsgChannel>> channels;

for (const auto& [name, config] : configs) {
    if (!config || !config->isValid()) {
        // 记录非法配置并按应用策略退出。
        continue;
    }

    auto channel = manager->createChannel(name, config, true);
    if (!channel) {
        // 记录打开失败及名称。
        continue;
    }
    channels.emplace(name, std::move(channel));
}

5. HybridChannel 开发​

优先使用 typed API：

  • 发送：HybridMsgChannel::send<T>；
  • 订阅：registerCallback<T>；
  • 类型、topic、QoS 和后端由部署 JSON 定义；
  • 应用编译对应消息头和 type support，但不直接编译或链接 RFramePlugin SDK。

推荐阅读：

  • src/channel/typed/typed_channel_example.cpp
  • src/channel/quickstart/channel_manager_example.cpp
  • src/channel/quickstart/hybrid_channel_example.cpp
  • src/channel/config/channel_config_json_example.cpp

先用 SHM-only 配置验证业务逻辑，再验证实际启用的 ROS 2 或 dual 通路。不要用 SHM 成功替代远端链路测试。

6. HybridRpcChannel 开发​

typed RPC 使用同一套请求和响应类型覆盖 SHM、ROS 2 或 dual 后端：

  • server 注册 registerHandler<Req, Resp>；
  • client 调用 call<Req, Resp>；
  • SHM 通路要求 client/server 的 service_prefix 与 method 分别一致；ROS 2 通路按最终服务全名匹配，无需采用相同的 prefix/method 拆分；双方请求/响应类型必须兼容，ROS 2 service type 必须一致（详见 §7.3.1）；
  • 调用方必须处理 timeout 和错误返回。

推荐阅读：

  • src/channel/typed/typed_rpc_example.cpp
  • src/rpc/hybrid_rpc_example.cpp

RFrameExample 运行方式：

# 终端 1
typed_rpc_example server --backend shm

# 终端 2
typed_rpc_example client --backend shm

运行命令用于 API 联调；二进制如何生成见《RFrame 编译指南》。

7. HybridChannel / HybridRpcChannel 用法清单（HMI 桥接 / 相机 / RobbyMotion）​

7.1. 文档定位​

本文是 RFrame 使用手册的上手清单，面向要按接口清单接业务的用户：

本文回答一个问题：拿到清单里的一条接口，怎么把它跑起来。每条路径都是"清单条目 → 可加载的 channels.json → 最小代码 → 验收命令"，复制即可用。

机制细节不在本文展开：

  • 编译接入（CMake、wheel、环境变量）见《RFrame 编译指南》；
  • API 与生命周期见本指南其余章节（§2–§6、§9）；
  • 部署、全局配置、JSON 字段规则见《RFrame 部署指南》。

接口事实（topic/service 名、类型、QoS、后端）一律以两份接口清单为准，本文不复制清单内容。

7.2. 五分钟上手路径​

  1. 按《RFrame 部署指南》§3 安装固定版本的 r-frame 和 r-frame-plugin。JSON 配置加载依赖 JsonDynamicDataPlugin：当前发布批次的 r-frame 默认 catalog（/opt/robby/r-frame/config/plugins.json）已启用它，插件 .so 由 r-frame-plugin 包提供。用以下命令确认环境就绪：
grep JsonDynamicDataPlugin /opt/robby/r-frame/config/plugins.json

无输出说明缺 r-frame-plugin、SDK 批次过旧或 catalog 被覆盖：loadFromFile() 会报 JSON plugin task 'FileToDynamicData::json' is unavailable 并返回空结果；

  1. 从清单选一条接口，在该清单附录中找到对应条目的 JSON 配置（§7.4）；
  2. 按场景复制最小代码：订阅 §7.5、发布 §7.6、调 service §7.7；
  3. 先配置检查，再实际收发。

7.3. 清单 JSON 与 loader 配置的关系​

清单附录中的 ```json 块就是 ChannelConfigLoader 可直接读取的最终配置（键为实例名，复制进 {"channels": {...}} 即可用）。附录只写与框架默认不同的字段；各字段的含义与生成规则如下：

字段配置中的位置说明
channel_type: "HYBRID"type: "hybrid"消息通道
channel_type: "HYBRID_RPC"type: "hybrid_rpc"RPC 通道
handler_type: "READER"/"WRITER"handler: "reader"/"writer"全小写
topictopic原样
enable_shmhybrid.enable_shm需要非默认 memory_size/queue_size/type_name 时必须显式提供 shm 子对象（相机条目必须）；缺省时框架用 channel 级 topic/handler 构造默认 SHM 配置
enable_ros2hybrid.enable_remote同一 ABI 字段的新名
remote_backendhybrid.remote_backendros2 / zenoh
auto_routehybrid.auto_route原样；不写 require_all（默认 false）等框架默认字段
memory_sizeshm.memory_size另补 shm.type_name；queue_size 默认 10，不写
type_nameshm.type_name、ros2.type_name两处一致
ros2_qos.depth/reliabilityros2.qos.depth/reliability只写非默认值；history/durability 等取框架默认
service拆分，见 §7.3.1不是原样进配置
service_typeros2.service_typesrv 全名
default_timeout_mshybrid_rpc.default_timeout_ms只写非默认值（默认 3000ms）；生效规则见 §7.7

Python 用户不经过 loader，把清单字段直接映射为 ChannelConfig 构造参数（见§8），字段名与扁平 JSON 基本一致。

7.3.1. service 条目必须拆分：service_prefix + method​

框架拼接 ROS 2 服务名的规则是（两侧斜杠会归一化）：

实际 ROS 2 服务名 = service_prefix 去尾斜杠 + "/" + method 去首斜杠

清单里的 service 全名（如 /gravity_calibration/calibrate）在配置中已拆成 service_prefix；调用/注册时 method 一律取全名最后一段。框架拼接 ROS 2 服务名时若把全名直接当 service_prefix，无论 method 传什么都拼不对（method 为空还会产生尾斜杠 /change_state/，ROS 2 直接拒绝）。拆分规则：

清单 service 全名service_prefixcall/registerHandler 的 method
/change_state/change_state
/gravity_calibration/calibrate/gravity_calibrationcalibrate
/kin_calib/begin/kin_calibbegin

hybrid_rpc.service_prefix 与 ros2.service_prefix 必须一致（《RFrame 部署指南》§6.4）。清单附录的配置已按此规则拆好，调用时 method 一律取服务全名的最后一段。

该拆分规则已过真实 ROS 2 端到端验证：上表前两种写法实际创建出的服务名分别为 /change_state 和 /gravity_calibration/calibrate（ros2 service list 外部确认，无双斜杠、无尾斜杠），各 40/40 次调用往返成功。

SHM 与 ROS 2 的匹配边界不同

  • SHM：prefix 决定传输通道，method 决定通道内的处理函数。 默认请求/响应 topic 分别为 service_prefix + "/req"、service_prefix + "/resp"；收到请求后，按请求帧中的 method 精确查找 handler。因此双方必须使用相同的 prefix 和相同的 method，仅拼接后的服务全名相同并不足以互通，也不会自动按完整服务名做最长 prefix 匹配。
  • ROS 2：按最终解析后的服务全名匹配。 在请求/响应 service type 一致的前提下，双方可以采用不同的 prefix/method 拆分，只要最终解析到同一个 ROS 2 服务全名即可。下面使用绝对名称，避免相对名称受 node namespace 影响。

例如同一个服务全名 /A/B/c_f：

配置service_prefixmethod默认 SHM 请求 topicROS 2 服务全名
拆分一/AB/c_f/A/req/A/B/c_f
拆分二/A/Bc_f/A/B/req/A/B/c_f

这两种拆分在 SHM 下属于不同的 RPC 通道：一端用拆分一、另一端用拆分二，无法通过 SHM 互通；在 ROS 2 下则可以匹配同一个服务。上文“method 取最后一段”是本清单统一采用的拆分约定，并非 ROS 2 要求双方必须采用这种拆分。

SHM 默认 topic 直接拼接后缀，不应套用 ROS 2 的斜杠归一化规则：例如 prefix /A/ 会生成 /A//req，建议非根 prefix 不带尾斜杠。启用 SHM + ROS 2 双通路时，应统一双方的 prefix/method 拆分，避免 ROS 2 可用而 SHM 不通。显式自定义 SHM transport topic 时，以实际 transport 配置为准。

7.3.2. 实例名规则​

channels 下的键是 ChannelManager 的实例名，不是 topic/service 名。清单附录的配置按"路径去斜杠转下划线"命名，service 条目加 _rpc 后缀：

清单条目实例名
/battery_state（订阅）battery_state
/initialpose（发布）initialpose
/change_state（service）change_state_rpc
/robby/sensors/camera/head/color/compressedrobby_sensors_camera_head_color_compressed
/robotic/human_robot_distance（同名 topic 同时订阅+发布）robotic_human_robot_distance_reader / robotic_human_robot_distance_writer

同一 topic 在同一份清单中既有订阅又有发布时（如 RobbyMotion 的 /robotic/human_robot_distance），实例名追加 handler 后缀消歧。

7.4. 从清单附录组装本进程的 channels.json​

三份清单文档的附录已按 §7.3 规则逐条给出可直接加载的 JSON 配置。框架不随 deb 分发汇总文件：按业务需要从清单附录复制对应条目，组装本进程自己的 config/channels.json（外层 {"channels": {...}}，键为实例名），再交给 ChannelConfigLoader::loadFromFile() / loadByName()。

清单条目数说明
RobbyHMI WS 桥接清单236 topic + 17 service；已按 nav/motion 范围清洗
RobbySensorCenter 相机清单12全部双通路（SHM + 远端）
RobbyMotion 清单9343 topic + 50 service；service 为 SHM+ROS 2 双通路，preferred_ingress: "ros2"

不要把多份清单的条目合并成一个 channels map 加载：不同清单对同名话题的通路配置可能不同（单通路 vs 双通路、QoS 差异），合并后后加载者会静默覆盖前者。各进程只加载自己那一份。

按名称取单条配置：

auto config = ChannelConfigLoader::loadByName(
    "config/channels.json", "battery_state");

7.5. 场景一：订阅 topic（HybridChannel reader）​

以订阅 /battery_state（sensor_msgs/msg/BatteryState，QoS best_effort）为例。

清单附录中复制出的条目（config/channels.json 由自己组装，见 §7.4）：

{
  "channels": {
    "battery_state": {
      "type": "hybrid",
      "topic": "/battery_state",
      "handler": "reader",
      "hybrid": {
        "enable_shm": false,
        "enable_remote": true,
        "auto_route": false
      },
      "ros2": {
        "type_name": "sensor_msgs/msg/BatteryState",
        "qos": {
          "reliability": "best_effort"
        }
      }
    }
  }
}

C++ 最小代码：

#include "r/frame/app/AppCore.h"
#include "r/frame/channel/ChannelConfigLoader.h"
#include "r/frame/channel/ChannelManager.h"
#include "r/frame/channel/HybridMsgChannel.h"

#include <sensor_msgs/msg/battery_state.hpp>

using namespace r::frame;

int main()
{
    auto app = AppCore::instance();
    if (app->init() != 0) {
        return 1;
    }
    app->addPlugin(ChannelManager::instance());

    auto config = ChannelConfigLoader::loadByName(
        "config/channels.json", "battery_state");
    if (!config || !config->isValid()) {
        app->shutdown();
        return 2;
    }

    auto channel = ChannelManager::instance()->createChannel(
        "battery_state", config, true);
    auto hybrid = std::dynamic_pointer_cast<HybridMsgChannel>(channel);
    if (!hybrid) {
        app->shutdown();
        return 3;
    }

    hybrid->registerCallback<sensor_msgs::msg::BatteryState>(
        [](std::shared_ptr<const sensor_msgs::msg::BatteryState>) {
            // 处理电量状态；回调内不要做无界阻塞工作。
        });

    // 运行业务……

    app->shutdown();   // Manager 托管的 channel 随 shutdown 统一清理
    return 0;
}

要点：

  • typed 回调签名是 std::function<void(std::shared_ptr<const T>)>；
  • QoS 必须与发布端兼容：清单标 best_effort 的 topic，订阅端配 reliable 会完全收不到数据；
  • 正常退出不需要逐个 close()/reset()（§9）；只有当局部 shared_ptr 的生命周期可能超过 AppCore::shutdown() 时（如静态/全局持有），才需要提前 reset()；
  • 完整可编译参考：/opt/robby/RFrameExample/src/channel/typed/typed_channel_example.cpp（deb 包 robby-r-frame-example）。

7.6. 场景二：发布 topic（HybridChannel writer）​

以发布 /initialpose（geometry_msgs/msg/PoseWithCovarianceStamped，QoS reliable）为例。配置条目从清单附录复制（handler: "writer"，其余结构与 §7.5 相同）；代码骨架同 §7.5 的 main()，差别只在创建后发送：

auto channel = ChannelManager::instance()->createChannel(
    "initialpose", config, true);
auto hybrid = std::dynamic_pointer_cast<HybridMsgChannel>(channel);

geometry_msgs::msg::PoseWithCovarianceStamped pose;
// 填充 pose ……
hybrid->send(pose);   // typed 发送，类型由配置 type_name 决定

要点：

  • 发送的 C++ 类型必须与配置里的 type_name 一致，否则序列化结果对端无法解析；
  • HMI 清单的 4 个发布 topic 均为 reliable，不要改成 best_effort。

7.7. 场景三：调用 service（HybridRpcChannel client）​

以调用 /change_state（state_machine_msgs/srv/ChangeState）为例。

清单附录中复制出的条目（注意 service_prefix 已按 §7.3.1 拆分）：

{
  "channels": {
    "change_state_rpc": {
      "type": "hybrid_rpc",
      "hybrid_rpc": {
        "service_prefix": "/",
        "enable_shm": false,
        "enable_ros2": true,
        "auto_route": false,
        "default_timeout_ms": 2000
      },
      "ros2": {
        "service_prefix": "/",
        "service_type": "state_machine_msgs/srv/ChangeState"
      }
    }
  }
}

C++ 最小代码（接 §7.5 的 main() 骨架，替换创建与调用部分）：

#include "r/frame/channel/HybridRpcChannel.h"
#include <state_machine_msgs/srv/change_state.hpp>

// init、loadByName("change_state_rpc") 同 §7.5。
auto channel = ChannelManager::instance()->createChannel(
    "change_state_rpc", config, true);
auto rpc = std::dynamic_pointer_cast<HybridRpcChannel>(channel);
if (!rpc) {
    app->shutdown();
    return 3;
}

state_machine_msgs::srv::ChangeState::Request req;
state_machine_msgs::srv::ChangeState::Response resp;
// 填充 req ……

// method 传服务全名的最后一段（§7.3.1）。
RpcStatus status = rpc->call("change_state", req, resp);
if (!status) {
    // 超时或错误：status.code / status.message。
}

要点：

  • typed 调用签名：call<TReq, TResp>(method, req, resp)，返回 RpcStatus（code == 0 为成功，可 if (!status) 判定）；
  • 必须处理 timeout 和错误返回。超时时间：不显式传 RpcCallOptions 时为框架默认 3000ms；传 RpcCallOptions 且 timeout_ms > 0 时按所传值；配置的 default_timeout_ms 只在显式传 timeout_ms <= 0 时兜底；
  • 请求/响应类型由 service_type 派生，client/server 两边必须一致；
  • 提供 server 端时对称使用 registerHandler<TReq, TResp>(method, handler)，handler 签名为 RpcStatus(const RpcContext&, const TReq&, TResp&)；
  • 完整可编译参考：/opt/robby/RFrameExample/src/channel/typed/typed_rpc_example.cpp（deb 包 robby-r-frame-example）。

7.8. 场景四：一份 JSON 批量加载多个 channel​

一个进程同时需要多个 channel 时（如下例：订阅电量、发布定位、调状态机 service），不必逐条 loadByName：loadFromFile 一次读出整份 JSON 的全部配置，批量创建后按实例名取用即可。

外部 JSON 形如——同一文件的 channels 下放多个条目，键即实例名（条目复制自 HMI 清单附录）：

{
  "channels": {
    "battery_state": {
      "type": "hybrid",
      "topic": "/battery_state",
      "handler": "reader",
      "hybrid": {
        "enable_shm": false,
        "enable_remote": true,
        "auto_route": false
      },
      "ros2": {
        "type_name": "sensor_msgs/msg/BatteryState",
        "qos": {
          "reliability": "best_effort"
        }
      }
    },
    "initialpose": {
      "type": "hybrid",
      "topic": "/initialpose",
      "handler": "writer",
      "hybrid": {
        "enable_shm": false,
        "enable_remote": true,
        "auto_route": false
      },
      "ros2": {
        "type_name": "geometry_msgs/msg/PoseWithCovarianceStamped"
      }
    },
    "change_state_rpc": {
      "type": "hybrid_rpc",
      "hybrid_rpc": {
        "service_prefix": "/",
        "enable_shm": false,
        "enable_ros2": true,
        "auto_route": false,
        "default_timeout_ms": 2000
      },
      "ros2": {
        "service_prefix": "/",
        "service_type": "state_machine_msgs/srv/ChangeState"
      }
    }
  }
}

C++ 代码：

#include "r/frame/app/AppCore.h"
#include "r/frame/channel/ChannelConfigLoader.h"
#include "r/frame/channel/ChannelManager.h"
#include "r/frame/channel/HybridMsgChannel.h"
#include "r/frame/channel/HybridRpcChannel.h"

#include <sensor_msgs/msg/battery_state.hpp>

#include <string>
#include <vector>

using namespace r::frame;

int main()
{
    auto app = AppCore::instance();
    if (app->init() != 0) {
        return 1;
    }
    app->addPlugin(ChannelManager::instance());

    // 一次加载整份 JSON，拿到全部 channel 配置
    auto configs = ChannelConfigLoader::loadFromFile(
        "config/channels.json");
    if (configs.empty()) {
        app->shutdown();
        return 2;
    }

    // 批量创建：实例名必须与 channels 下的键一致；任一失败记录并退出，不静默跳过
    auto manager = ChannelManager::instance();
    const std::vector<std::string> needed = {
        "battery_state", "initialpose", "change_state_rpc"};
    for (const auto& name : needed) {
        auto it = configs.find(name);
        if (it == configs.end() || !it->second || !it->second->isValid()) {
            app->shutdown();
            return 3;
        }
        if (!manager->createChannel(name, it->second, true)) {
            app->shutdown();
            return 4;
        }
    }

    // 创建后由 ChannelManager 托管，按实例名取用并按类型 cast
    auto battery = std::dynamic_pointer_cast<HybridMsgChannel>(
        manager->getChannel("battery_state"));
    auto rpc = std::dynamic_pointer_cast<HybridRpcChannel>(
        manager->getChannel("change_state_rpc"));
    if (!battery || !rpc) {
        app->shutdown();
        return 5;
    }

    battery->registerCallback<sensor_msgs::msg::BatteryState>(
        [](std::shared_ptr<const sensor_msgs::msg::BatteryState>) {
            // 处理电量状态
        });
    // rpc->call("change_state", req, resp); method 取服务全名最后一段（§7.3.1）

    app->shutdown();  // Manager 托管的 channel 随 shutdown 统一关闭（§9）
    return 0;
}

要点：

  • loadFromFile 返回整份文件的 name -> config map，批量创建时逐个查找并检查 isValid()，失败不静默（§4）；
  • 创建后的 channel 由 ChannelManager 托管，之后随时可用 getChannel(name) 取用，正常退出只需 app->shutdown()（§9）；
  • 仍遵守 §7.4 的提醒：多份清单的 channels.json 不要合并加载，各进程只加载自己那一份。

7.9. 双通路（SHM + ROS 2 / zenoh）​

相机清单的 12 个话题全部是双通路：同机走 SHM（低延迟大图），跨机走远端。以清单附录的头相机压缩图条目为例：

"robby_sensors_camera_head_color_compressed": {
  "type": "hybrid",
  "topic": "/robby/sensors/camera/head/color/compressed",
  "handler": "reader",
  "hybrid": {
    "enable_shm": true,
    "enable_remote": true,
    "auto_route": false
  },
  "shm": {
    "memory_size": 5242880,
    "type_name": "sensor_msgs/msg/CompressedImage"
  },
  "ros2": { "type_name": "...", "qos": { "reliability": "best_effort" } }
}

规则：

  • 双通路去重是框架默认行为：Discovery 确认双通路都可达后，只走 SHM 一路，不再重复投递；注意发现尚未收敛的 bootstrap 阶段两路会乐观启用，这段时间仍可能收到重复数据，业务回调要容忍；
  • 任一后端打开失败不影响另一路（require_all 默认 false；改为 true 则整体失败）；
  • 代码与 §7.5 完全相同——双通路对业务代码透明，差别只在配置。

/robby/sensors/camera/head/depth/compressed 的远端后端是 zenoh：配置写 hybrid.remote_backend: "zenoh"，不提供 ros2 子对象，QoS 默认 best_effort 即对应清单意图。可选 zenoh 子对象还能覆盖 key_expr、会话参数（mode/connect/listen）等细节；缺省配置即可合法打开，跨机连通取决于 zenoh 发现/会话配置。后端插件需求：zenoh 后端内建于 r-frame 本体，无需额外插件；ROS 2 远端后端由 r-frame-plugin 的 ros2.runtime 插件提供（《RFrame 部署指南》§3.2）。

先用 SHM-only 验证业务逻辑，再单独验证远端通路；SHM 跑通不代表 ROS 2 / zenoh 通路通（§5）。

7.10. 常见坑​

现象原因处理
订阅完全收不到数据QoS 与对端不兼容（reliable 订阅配 best_effort 发布）严格按清单附录的 QoS 配置
SHM 通路收发失败、大图丢帧memory_size 小于一帧按清单值配置（相机为 5242880），不要拍脑袋改小
service 调用报服务不存在 / 名字非法service 全名直接当了 service_prefix按 §7.3.1 拆分 prefix + method
ROS 2 RPC 可用，但 SHM RPC 不通双方服务全名相同，但 service_prefix / method 拆分不同，落到了不同 SHM topic对齐双方 prefix 和 method，按 §7.3.1 检查实际 SHM 请求/响应 topic
RPC 超时时间与预期不符不传 RpcCallOptions 时按框架默认 3000ms，而非配置的 default_timeout_ms见 §7.7 超时规则
registerHandler 返回 false、RPC server 起不来同一 service_prefix 已有在役 server（同进程或另一进程）；SHM 通路按 prefix 独占注册日志含 refusing SHM handler registration: prefix='...' is already served；换不同 prefix 或先下线旧 server（旧进程已退出时租约会经存活检查后自动释放）
SHM 通路 RPC 大报文调用失败默认 shm_buffer_size 为 65536（64KB），单条请求/响应需放进该缓冲报文超过 64KB 时在 hybrid_rpc.shm_buffer_size 按最大报文上调
配置加载结果为空，日志报 FileToDynamicData::json is unavailable未装 r-frame-plugin（提供 JsonDynamicDataPlugin）；也可能是 SDK 批次过旧、缺 {"channels": {...}} 外层、JSON 语法错按《RFrame 部署指南》§3.2 安装 r-frame-plugin；再用 channel_config_json_example 检查
运行期莫名崩溃或 RPC 静默超时插件 .so 与 RFrame 库不是同一 SDK 批次按《RFrame 部署指南》§3.2 钉版本，不混装
新配置不生效的旧字段消息 Channel 写了兼容字段 hybrid.dedup新配置不写；enable_remote 取代 enable_ros2（hybrid_rpc 内仍用 enable_ros2）
虚拟环境 import 到别的 rframewheel 与 native deb 不同批次《RFrame 编译指南》§3.4 版本一致性检查

8. Python 用法​

8.1. 初始化和日志​

from rframe import App, Log

app = App()
app.init()

Log.info("RFrame Python application started")

app.shutdown()

全局日志级别由 rframe.json 或 RFRAME_LOG_LEVEL 控制，部署方式见《RFrame 部署指南》。

8.2. Hybrid Topic​

from std_msgs.msg import String
from rframe import App, ChannelConfig, ChannelType, HandlerType, HybridChannel

app = App()
app.init()

config = ChannelConfig(
    topic="/rframe/python/demo",
    handler_type=HandlerType.WRITER,
    channel_type=ChannelType.HYBRID,
    enable_shm=True,
    enable_ros2=False,
    auto_route=False,
    require_all=True,
    memory_size=64 * 1024,
    type_name="std_msgs/msg/String",
)
channel = HybridChannel(config)

message = String()
message.data = "hello from Python"
channel.send(message)

channel.close()
app.shutdown()

订阅端使用 typed callback：

def on_message(message: String):
    print(message.data, flush=True)

channel.register_typed_callback(String, on_message)

完整示例：python/typed_channel_interop.py。

8.3. Hybrid RPC​

from pathlib import Path
import json

from rframe import App, ChannelConfig, ChannelType
from rframe.channel import ChannelManager

app = App()
app.init()
app.add_plugin("ChannelManager")

document = json.loads(Path("config/interop_shm.json").read_text())
entry = document["channels"]["interop"]
rpc = entry["hybrid_rpc"]

config = ChannelConfig(
    topic=rpc["service_prefix"],
    channel_type=ChannelType.HYBRID_RPC,
    enable_shm=rpc.get("enable_shm", True),
    enable_ros2=rpc.get("enable_ros2", False),
    auto_route=rpc.get("auto_route", False),
    require_all=rpc.get("require_all", False),
    memory_size=rpc.get("shm_buffer_size", 65536),
)
channel = ChannelManager.create("interop", config=config)

response = channel.call("Method", b"request", timeout_ms=3000)

app.shutdown()

服务端通过 register_handler(method, handler) 注册处理函数。typed CDR RPC 的完整 Python/C++ 互通示例见 python/rpc_interop.py。

Python binding 使用 ChannelConfig 创建配置；当配置来自 JSON 时，业务读取所需条目并映射为 ChannelConfig。不要假设 Python 暴露了与 C++ ChannelConfigLoader 完全相同的快捷接口。

9. 生命周期管理​

通过 ChannelManager 创建并登记的 Channel 由 Manager 统一管理。应用正常退出时，ChannelManager::shutdown() 会清理并关闭其持有的全部 Channel；当 ChannelManager 已交给 AppCore 管理时，业务通常只需执行：

app->shutdown();

不要求业务在正常退出路径中逐个执行 close()、removeChannel() 和 reset()。

以下场景才需要显式移除：

  • 应用运行期间提前下线某个 Channel；
  • 使用同名配置重建 Channel；
  • 希望立即撤销发现信息并确定性释放该 Channel 的资源。

此时调用 Manager 的移除接口即可，removeChannel() 会负责 unannounce、解绑并关闭 Channel：

manager->removeChannel(channel_name);

只有绕过 ChannelManager 直接构造和持有的 Channel，才由业务负责在不用时显式 close()。