Skip to main content

RFrame Development Guide

Draft for website testing. This is a condensed English draft, not a complete translation. Both language versions await final content-owner updates. Original technical details, code comments, tables, and illustrations are retained in Chinese in the expandable reference below.

Read the Chinese source

1. Scope and workspace​

This guide covers AppCore, ChannelConfigLoader, ChannelManager, typed HybridChannel, typed HybridRpcChannel, and lifecycle handling. Start with Getting Started with RFrame; see the Build Guide for compilation and the Deployment Guide for configuration.

Installed examples from robby-r-frame-example are under /opt/robby/RFrameExample/. In the supplied development container, /mnt/ssd is persistent. Create /mnt/ssd/workspace for source, configuration, builds, logs, and data. Container restarts remove user changes outside /mnt/ssd; copy examples into the persistent workspace before editing.

2. Initialize the application​

Call AppCore::init() before other RFrame operations. Set up application resources and logging before starting communication. Original C++ initialization and logging examples are retained in the appendix.

3. Load configuration​

ChannelConfigLoader supports loading from a standalone file, a string, or the global configuration. Channels can also be configured in C++ without JSON. Preserve configuration field names and values when adapting the examples.

4. Create channels​

Use ChannelManager to create and register channels. Keep instance names consistent between configuration and code. The manager handles the registered channels' lifecycle.

5. Hybrid topic channels​

Use a typed HybridChannel reader to subscribe and a writer to publish. Select the correct message type and configure the required local and remote transports. The original reader, writer, and callback examples are available below.

6. Hybrid RPC channels​

Use HybridRpcChannel for typed request/response communication. Keep the service prefix and method separate as required by the loader configuration. Check the result and error handling shown in the examples.

7. Interface configurations and scenarios​

The interface lists provide loader-ready JSON entries keyed by instance name. Copy selected entries into your process's channels object. Do not add unspecified fields merely because another interface uses them; omitted fields take framework defaults.

ScenarioReference
State and core functionsCore Function Interfaces
NavigationNavigation Interfaces
Camera streamsSensor Camera Topics
Other sensorsSensor Functions
Motion controlMotion Control Interfaces

The detailed examples cover topic subscription, publication, service calls, loading several channels from one JSON file, and dual transports using SHM with ROS 2 or zenoh. Check the instance-name rules and common pitfalls in the source appendix.

8. Python​

The Python section demonstrates initialization and logging, Hybrid Topic communication, and Hybrid RPC calls. The original examples are preserved, including imports, configuration keys, and cleanup calls.

9. Lifecycle management​

When ChannelManager is managed by AppCore, normal shutdown generally requires only app->shutdown(). The manager closes its registered channels; application code need not individually close, remove, and reset each one.

Call manager->removeChannel(channel_name) when removing a channel early, recreating a channel with the same name, or immediately withdrawing discovery information and releasing resources. Channels created and held directly outside the manager must be closed explicitly by their owner.

10. Original technical reference (Chinese)​

Expand the original chapters, tables, code, and illustrations

文档总入口：《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 编译指南》。

用户开发目录与持久化​

当前提供的开发容器中，/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()。