3. 通道 Channel

Channel 是 autolink 的 持续数据流 抽象，对标 ROS 2 topic。Writer 发布、Reader 订阅；匹配条件为 channel 名一致 + 消息类型一致。

本文 §3	相关文档
发布订阅	§0 指南 · §2 节点 Node · §10 组件 Component · §12 调度 Scheduler

3.1 传输与发现

场景	传输
同进程	INTRA（内存直传）
同机多进程	SHM（共享内存）
跨主机	RTPS（Fast DDS）

发现经 UDP 多播，无 central master；匹配完成后数据不经发现通道。

3.2 消息类型

message Chatter {
  uint64 timestamp = 1;
  uint64 lidar_timestamp = 2;
  uint64 seq = 3;
  bytes content = 4;
}

Autonomy 业务消息为 autonomy::commsgs::* struct，跨进程边界 ToProto() 后走 Channel。亦支持 POD 包装类型（见 pod_talker_listener.cpp + pod_packet.hpp），适合高频小结构体。

3.3 发布与订阅

发布（talker.cpp）：

auto node = autolink::CreateNode("talker");
auto writer = node->CreateWriter<Chatter>("channel/chatter");

while (autolink::OK()) {
    auto msg = std::make_shared<Chatter>();
    msg->set_seq(seq++);
    msg->set_content("Hello, autolink!");
    writer->Write(msg);   // 同步写入传输层
    rate.Sleep();
}

订阅（listener.cpp）：

void MessageCallback(const std::shared_ptr<Chatter>& msg) {
    AINFO << "Received message seq-> " << msg->seq();
    AINFO << "msgcontent->" << msg->content();
}

auto reader = node->CreateReader<Chatter>("channel/chatter", MessageCallback);

回调在 Scheduler 线程执行，勿长时间阻塞；重逻辑移交工作线程或 Component Proc。

3.4 ReaderConfig 与 QoS

ReaderConfig 默认值（node_channel_impl.hpp）：

字段	默认	含义
`qos_profile.history`	`KEEP_LAST`	只保留最近 N 条
`qos_profile.depth`	`1`	历史深度
`qos_profile.reliability`	`RELIABLE`	可靠传输
`qos_profile.durability`	`VOLATILE`	不保留晚加入订阅者之前的数据
`pending_queue_size`	`DEFAULT_PENDING_QUEUE_SIZE`	待处理回调队列长度

控制 / 规划环建议 只处理最新一帧：

autolink::ReaderConfig cfg;
cfg.channel_name = "/plan";
cfg.pending_queue_size = 1;
node->CreateReader<Path>(cfg, [](const std::shared_ptr<Path>& p) {
    // 轻量处理
});

队列满时旧消息被丢弃，避免延迟累积。

亦可通过 proto::RoleAttributes 精细设置 channel 名与 QoS：

proto::RoleAttributes attr;
attr.set_channel_name("/sensor/imu");
attr.mutable_qos_profile()->set_depth(10);
node->CreateWriter<Imu>(attr);

3.5 POD 消息

pod_talker_listener.cpp 演示四步：

定义 trivially copyable POD 结构体
包装类实现 TypeName、SerializeToArray / ParseFromArray
CreateWriter<PodPacket> / CreateReader<PodPacket>
填充字段后 Write

auto writer = node->CreateWriter<PodPacket>("channel/pod_demo");
auto reader = node->CreateReader<PodPacket>("channel/pod_demo", OnPodMessage);

auto msg = std::make_shared<PodPacket>();
msg->pod.seq = seq;
msg->pod.value = static_cast<double>(seq) * 0.5;
msg->pod.timestamp_ns = Time::Now().ToNanosecond();
writer->Write(msg);

单进程验证：./autolink_example_pod_talker_listener；跨进程：pod_talker + pod_listener。

3.6 Rate 定频

C++ Rate（Hz）：

autolink::Rate rate(1.0);
while (autolink::OK()) {
    writer->Write(msg);
    rate.Sleep();
}

Python（py_talker.py）：

rate = autolink_time.Rate(1.0)
while not autolink.is_shutdown():
    writer.write(msg)
    rate.sleep()

3.7 数据路径

发现匹配完成后，业务数据不经发现通道，按拓扑自动选择传输后端。

Channel Write → 传输 → pending_queue → 调度 → 回调与 §1.2 内部分层 L3–L4 对应；Component 在回调处接入 DataVisitor

发布 Writer::Write

INTRA · SHM · RTPS

订阅 pending_queue

Scheduler

回调 / DataVisitor

模式	回调目标
Binary	用户 `CreateReader` 回调
Component	`DataVisitor` → 多路齐备 → `Proc`（[§1.4](01_architecture.md#14-component-数据流)）

回调在 Scheduler 工作线程执行，保持轻量；重逻辑移交业务线程。

3.8 Python API

发布（py_talker.py）：

test_node = autolink.Node("node_name1")
writer = test_node.create_writer("channel/chatter", Chatter, 6)
rate = autolink_time.Rate(1.0)
while not autolink.is_shutdown():
    msg = Chatter()
    msg.seq = g_count
    msg.content = b"I am python talker."
    writer.write(msg)
    g_count += 1
    rate.sleep()

订阅（py_listener.py）：

def callback(data):
    print("seq:", data.seq, "content:", data.content)

test_node = autolink.Node("listener")
test_node.create_reader("channel/chatter", Chatter, callback)
test_node.spin()

3.9 CLI 工具

autolink_channel list          # 列出活跃 channel 与类型
autolink_monitor               # 实时消息监控
autolink_recorder record -a -o x.record   # 录包
autolink_recorder play -f x.record        # 回放

录包示例源码：autolink/examples/cpp/record.cpp。

导航：← §2 节点 Node · §0 指南 · §4 服务 Service →