refactor: forward shared topics from callbacks

Author: Jiu-xiao

README.md

@@ -13,9 +13,26 @@ SharedTopicClient is a client module for multi-topic data sharing and transparen
 ## 构造参数 / Constructor Arguments
 
 - uart_name: 串口设备名 / UART device name (e.g., "uart_cdc")
-- task_stack_depth: 任务堆栈大小 / Task stack depth (e.g., 512)
-- buffer_size: 发送缓冲区字节数 / TX buffer size (e.g., 256)
-- topic_names: 需要订阅并转发的 Topic 名称列表 / List of topic names to subscribe and forward (e.g., ["topic1", "topic2"])
+- buffer_size: 单个 Topic 串口包最大字节数 / Maximum bytes of one forwarded Topic packet (e.g., 256)
+- topic_configs: 需要订阅并转发的 Topic 配置列表。每项可以只写 topic 名，也可以写
+  `[topic, domain]`。/ Topic configs to subscribe and forward. Each item may be a
+  topic name or `[topic, domain]`.
+
+## 运行方式
+
+`SharedTopicClient` 不创建发送线程。模块注册 Topic callback；每次 Topic 发布时，
+callback 内直接完成打包并把当前包写入 UART `write_port`。UART 写入仍走 libxr
+非阻塞写队列，包内容在 `Write()` 返回前已经复制到 UART 队列。
+
+## Timestamp
+
+`SharedTopicClient` 转发 Topic 时会保留 libxr message envelope timestamp：
+
+1. 本地 Topic callback 收到 `(timestamp, payload)`。
+2. `Topic::PackData(topic_crc, buffer, timestamp, payload)` 写入串口包。
+3. 对端 `SharedTopic` 解析后用同一个 timestamp 发布到对端 domain。
+
+因此同步类 topic 不需要在 payload 里重复携带时间戳；payload 只保留业务字段即可。
 
 ## 依赖 / Depends
 

SharedTopicClient.hpp

@@ -5,7 +5,6 @@
 module_description: SharedTopicClient 是一个多 Topic 数据共享与串口转发客户端模块。它用于通过 UART 将多个 Topic 的数据统一打包、发送，实现消息流的串口透明同步转发，适用于分布式系统的多主题数据同步或边缘数据采集。 / SharedTopicClient is a client module for multi-topic data sharing and transparent UART forwarding. It subscribes to multiple Topics, packs their updates, and transmits them via UART, enabling efficient and reliable message synchronization over serial connections—ideal for distributed systems or edge data acquisition.
 constructor_args:
   - uart_name: "uart_cdc"
-  - task_stack_depth: 2048
   - buffer_size: 256
   - topic_configs:
     - "topic1"
@@ -16,16 +15,20 @@ depends: []
 === END MANIFEST === */
 // clang-format on
 
+#include <cstddef>
+#include <cstdint>
+
 #include "app_framework.hpp"
+#include "message.hpp"
 #include "uart.hpp"
 
 class SharedTopicClient : public LibXR::Application {
  public:
-  typedef struct {
+  struct CallbackInfo {
     SharedTopicClient* client;
     uint32_t topic_crc32;
     uint32_t index;
-  } CallbackInfo;
+  };
 
   struct TopicConfig {
     const char* name;
@@ -39,16 +42,16 @@ class SharedTopicClient : public LibXR::Application {
 
   SharedTopicClient(LibXR::HardwareContainer& hw,
                     LibXR::ApplicationManager& app, const char* uart_name,
-                    uint32_t task_stack_depth, uint32_t buffer_size,
+                    uint32_t buffer_size,
                     std::initializer_list<TopicConfig> topic_configs)
-      : uart_(hw.template Find<LibXR::UART>(uart_name)),
-        tx_buffer_(new uint8_t[buffer_size], buffer_size),
-        tx_queue_(buffer_size) {
+      : uart_(hw.template Find<LibXR::UART>(uart_name)) {
     ASSERT(uart_ != nullptr);
+    ASSERT(uart_->write_port_ != nullptr);
+    ASSERT(uart_->write_port_->queue_data_ != nullptr);
 
     topics_pack_buffer_ = new LibXR::RawData[topic_configs.size()];
-
     uint32_t i = 0;
+    size_t max_packet_size = 0;
 
     for (auto config : topic_configs) {
       auto domain = LibXR::Topic::Domain(config.domain);
@@ -57,21 +60,26 @@ class SharedTopicClient : public LibXR::Application {
         XR_LOG_ERROR("Topic not found: %s/%s", config.domain, config.name);
         ASSERT(false);
       }
+      const size_t packet_size =
+          ans->data_.max_length + LibXR::Topic::PACK_BASE_SIZE;
+      ASSERT(packet_size <= buffer_size);
+      max_packet_size = LibXR::max(max_packet_size, packet_size);
       topics_pack_buffer_[i] = LibXR::RawData(
-          new uint8_t[ans->data_.max_length + LibXR::Topic::PACK_BASE_SIZE],
-          ans->data_.max_length + LibXR::Topic::PACK_BASE_SIZE);
+          new uint8_t[packet_size], packet_size);
+
+      void (*func)(bool, CallbackInfo, LibXR::MicrosecondTimestamp,
+                   LibXR::ConstRawData&) =
+          [](bool in_isr, CallbackInfo info,
+             LibXR::MicrosecondTimestamp timestamp, LibXR::ConstRawData& data) {
+            auto& buffer = info.client->topics_pack_buffer_[info.index];
+            ASSERT(data.size_ + LibXR::Topic::PACK_BASE_SIZE <= buffer.size_);
+            LibXR::Topic::PackData(info.topic_crc32, buffer, timestamp, data);
 
-      void (*func)(bool, CallbackInfo, LibXR::RawData&) =
-          [](bool in_isr, CallbackInfo info, LibXR::RawData& data) {
             LibXR::WriteOperation op;
-            LibXR::Topic::PackData(info.topic_crc32,
-                                   info.client->topics_pack_buffer_[info.index],
-                                   data);
-            info.client->tx_queue_.PushBatch(
-                static_cast<uint8_t*>(
-                    info.client->topics_pack_buffer_[info.index].addr_),
-                info.client->topics_pack_buffer_[info.index].size_);
-            info.client->tx_sem_.PostFromCallback(in_isr);
+            auto ans = info.client->uart_->Write(
+                {buffer.addr_, data.size_ + LibXR::Topic::PACK_BASE_SIZE}, op,
+                in_isr);
+            UNUSED(ans);
           };
 
       auto msg_cb = LibXR::Topic::Callback::Create(
@@ -84,35 +92,14 @@ class SharedTopicClient : public LibXR::Application {
       i++;
     }
 
-    tx_thread_.Create(this, TxThreadFun, "SharedTopicClientTxThread",
-                      task_stack_depth, LibXR::Thread::Priority::REALTIME);
+    ASSERT(max_packet_size <= uart_->write_port_->queue_data_->MaxSize());
 
     app.Register(*this);
   }
 
-  static void TxThreadFun(SharedTopicClient* client) {
-    LibXR::Semaphore write_op_sem;
-    LibXR::WriteOperation op(write_op_sem);
-    LibXR::WriteOperation op_none;
-    while (true) {
-      client->tx_sem_.Wait();
-      auto size =
-          LibXR::min(client->tx_queue_.Size(), client->tx_buffer_.size_);
-      if (size > 0 && client->tx_queue_.PopBatch(
-                          static_cast<uint8_t*>(client->tx_buffer_.addr_),
-                          size) == LibXR::ErrorCode::OK) {
-        client->uart_->Write(
-            {static_cast<uint8_t*>(client->tx_buffer_.addr_), size}, op_none);
-      }
-    }
-  }
   void OnMonitor() override {}
 
  private:
   LibXR::UART* uart_;
-  LibXR::RawData tx_buffer_;
-  LibXR::LockFreeQueue<uint8_t> tx_queue_;
   LibXR::RawData* topics_pack_buffer_;
-  LibXR::Semaphore tx_sem_;
-  LibXR::Thread tx_thread_;
 };