devicesbus modbus rtu

CMakeLists.txt

@@ -119,16 +119,37 @@ qt_add_executable(softbus_daemon
     src/core/plugin_system/IProtocolPlugin.h
     src/core/plugin_system/PluginManager.h
     src/core/plugin_system/PluginManager.cpp
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp
+    src/core/models/Types.h
+    src/core/models/MetadataDef.h
+    src/core/models/DataMapping.h
+    src/core/models/DOMMessage.h
+    src/core/metadata/MetadataRegistry.h
+    src/core/metadata/MetadataRegistry.cpp
+    src/core/metadata/ProfileRegistry.h
+    src/core/metadata/ProfileRegistry.cpp
 
     # message bus pipeline
     src/message_bus/pipeline/PayloadRef.h
     src/message_bus/pipeline/PipelineContext.h
     src/message_bus/pipeline/PipelineEngine.h
     src/message_bus/pipeline/PipelineEngine.cpp
+    src/message_bus/pipeline/protocol/ProtocolEnvelope.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h
+    src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp
     src/message_bus/pipeline/stages/1_framers/IFramer.h
-    src/message_bus/pipeline/stages/1_framers/DefaultFramer.cpp
+    src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h
+    src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp
+    src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h
+    src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp
+    src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h
+    src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp
     src/message_bus/pipeline/stages/2_parsers/IParserStage.h
     src/message_bus/pipeline/stages/3_filters/IFilterStage.h
+    src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h
+    src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp
     src/message_bus/pipeline/stages/4_validators/IValidator.h
 
     # message bus ingress/egress
@@ -179,6 +200,52 @@ endif()
 
 add_subdirectory(plugins)
 
+option(SOFTBUS_BUILD_TESTS "Build unit tests" OFF)
+if(SOFTBUS_BUILD_TESTS)
+    find_package(Qt6 6.5 REQUIRED COMPONENTS Test)
+    qt_add_executable(softbus_modbus_rtu_framer_test
+        tests/modbus_rtu_framer_test.cpp
+        src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp
+        src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp
+    )
+    target_include_directories(softbus_modbus_rtu_framer_test PRIVATE
+        ${CMAKE_SOURCE_DIR}
+        ${CMAKE_SOURCE_DIR}/src
+    )
+    target_link_libraries(softbus_modbus_rtu_framer_test PRIVATE
+        Qt6::Core
+        Qt6::Test
+    )
+
+    qt_add_executable(softbus_protocol_parse_filter_test
+        tests/protocol_parse_filter_test.cpp
+        src/core/metadata/MetadataRegistry.cpp
+        src/core/metadata/ProfileRegistry.cpp
+        src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp
+    )
+
+    qt_add_executable(softbus_modbus_rtu_pdu_codec_test
+        tests/modbus_rtu_pdu_codec_test.cpp
+        src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp
+        src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp
+    )
+    target_include_directories(softbus_modbus_rtu_pdu_codec_test PRIVATE
+        ${CMAKE_SOURCE_DIR}
+        ${CMAKE_SOURCE_DIR}/src
+    )
+    target_link_libraries(softbus_modbus_rtu_pdu_codec_test PRIVATE
+        Qt6::Core
+        Qt6::Test
+    )
+    target_include_directories(softbus_protocol_parse_filter_test PRIVATE
+        ${CMAKE_SOURCE_DIR}
+        ${CMAKE_SOURCE_DIR}/src
+    )
+    target_link_libraries(softbus_protocol_parse_filter_test PRIVATE
+        Qt6::Core
+        Qt6::Test
+    )
+endif()
 
 install(TARGETS softbus_daemon
     BUNDLE  DESTINATION .

config/daemon_config.json

@@ -1,5 +1,14 @@
 {
   "plugin_dir": "bin/plugins",
+  "pipeline": {
+    "parallelPipeline": false,
+    "parseWorkerCount": 2,
+    "orderedEgress": true,
+    "framedQueueMax": 4096,
+    "maxReorderDepth": 256,
+    "modbusRtuMaxFrameBytes": 256,
+    "modbusRtuInterFrameTimeoutMs": 0
+  },
   "device_tree": [
     {
       "id": "serial_1",
@@ -9,6 +18,7 @@
       "driver": "hardware/drivers/serial/SerialQtDriver",
       "capabilitiesRef": "src/hardware/drivers/serial/parameters.json",
       "params": {
+        "protocol": "modbus_rtu",
         "baudRate": 115200,
         "dataBits": 8,
         "parity": "none",

plugins/protocols/modbus_rtu/plugin_stub.cpp

@@ -1 +1,2 @@
-extern "C" void softbus_modbus_rtu_plugin_stub() {}
+extern "C" void
+softbus_modbus_rtu_plugin_stub() {}

src/core/memory/MemoryPool.h

@@ -59,35 +59,7 @@ public:
             this->release(*b);
             delete b;
         });
-    BlockPtr allocate(std::size_t requiredSize)
-    {
-        if (requiredSize > m_blockSize) {
-            return {};
-        }
-
-        std::size_t slot = 0;
-        {
-            std::lock_guard<std::mutex> lk(m_mutex);
-            if (m_freeSlots.empty()) {
-                return {};
-            }
-            slot = m_freeSlots.front();
-            m_freeSlots.pop();
-            m_inUse[slot] = true;
-        }
-
-        auto* raw = new PoolBlock{
-            m_storage.data() + slot * m_blockSize,
-            m_blockSize,
-            requiredSize,
-            slot,
-        };
-        return BlockPtr(raw, [this](PoolBlock* b) {
-            if (!b) return;
-            this->release(*b);
-            delete b;
-        });
-
+    }
 
     void release(const PoolBlock& block)
     {

src/device_bus/manager/DeviceBusManager.cpp

@@ -10,6 +10,7 @@
 #include "message_bus/ingress/DriverIngressAdapter.h"
 #include "message_bus/pipeline/PipelineEngine.h"
 #include "core/plugin_system/PluginManager.h"
+#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h"
 #include "device_bus/manager/SerialDeviceManager.h"
 #include "devices/DeviceTypes.h"
 #include "utils/logs/logging.h"
@@ -56,7 +57,10 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig)
     if (!m_pluginManager) {
         m_pluginManager = std::make_shared<softbus::core::plugin_system::PluginManager>();
     }
+    m_pluginManager->registerProtocolPlugin(softbus::core::plugin_system::makeModbusRtuProtocolPlugin());
     if (m_pipelineEngine) {
+        m_pipelineEngine->setMemoryPool(m_memoryPool);
+        m_pipelineEngine->loadRuntimeConfigFromJson(rootConfig.value(QStringLiteral("pipeline")).toObject());
         m_pipelineEngine->setPluginManager(m_pluginManager);
         m_pipelineEngine->start();
     }

src/device_bus/manager/SerialDeviceManager.cpp

@@ -128,9 +128,10 @@ bool initializeSerialDeviceWithDeps(const QString& endpoint, const QJsonObject&
 
     softbus::devices::DeviceInfo info;
     info.id = -1;
-    info.endpoint = endpoint;
-    info.type = QStringLiteral("serial");
-    info.kind = softbus::devices::DeviceKind::Serial;
+    info.endpoint  = endpoint;
+    info.type      = QStringLiteral("serial");
+    info.kind      = softbus::devices::DeviceKind::Serial;
+    info.protocol  = params.value(QStringLiteral("protocol")).toString(QStringLiteral("modbus_rtu"));
 
     softbus::devices::physical::SerialDevice::Config cfg;
     cfg.endpoint = endpoint;

src/devices/physical/SerialDevice.cpp

@@ -114,8 +114,10 @@ void SerialDevice::onRead(const std::vector<std::uint8_t>& data)
     LOG_INFO() << "SerialDevice: onRead data size" << data.size() << "endpoint" << m_config.endpoint;
     // 将数据推送到 ingress 队列 包含端口名、协议、数据、时间戳、traceId
     softbus::message_bus::pipeline::PipelineContext ctx;
+    ctx.direction = softbus::message_bus::pipeline::BusDirection::Upstream;
     ctx.endpoint = m_config.endpoint;
     ctx.protocolHint = m_info.protocol;
+    ctx.frameKind    = softbus::message_bus::pipeline::FrameKind::StreamChunk;
     ctx.payload.block = std::move(block);
     ctx.payload.length = data.size();
     ctx.payloadSize = data.size();

src/message_bus/pipeline/PipelineContext.h

@@ -5,14 +5,31 @@
 #include <cstdint>
 
 #include <QDateTime>
+#include <QJsonObject>
 #include <QMetaType>
 #include <QString>
+#include <QVector>
 
+#include "core/models/DOMMessage.h"
 #include "message_bus/pipeline/PayloadRef.h"
+#include "message_bus/pipeline/protocol/ProtocolEnvelope.h"
 
 namespace softbus::message_bus::pipeline
 {
 
+enum class FrameKind
+{
+    StreamChunk,
+    CompleteFrame,
+};
+
+enum class BusDirection
+{
+    Unknown,
+    Upstream,
+    Downstream,
+};
+
 struct PipelineContext
 {
     int deviceId{-1}; // 设备ID
@@ -24,6 +41,18 @@ struct PipelineContext
 
     PayloadRef payload; // 有效载荷
     std::size_t payloadSize{0}; // 有效载荷大小
+
+    FrameKind frameKind{FrameKind::StreamChunk};
+    BusDirection direction{BusDirection::Unknown};
+    std::uint64_t frameSeq{0};
+
+    /// 强类型协议解析结果（推荐下游使用）；多协议扩展时只增不改 variant。
+    softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol;
+
+    /// 兼容旧逻辑与日志；新代码请优先使用 `protocol`，勿依赖此处 JSON 字段名。
+    QJsonObject parsed;
+    QVector<softbus::core::models::DOMMessage> domMessages;
+    QString framerError; // 成帧/Pipeline 旁路诊断（可选）
 };
 
 } // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/PipelineEngine.cpp

@@ -1,14 +1,72 @@
 #include "message_bus/pipeline/PipelineEngine.h"
 
+#include <algorithm>
+#include <iomanip>
+#include <sstream>
 #include <chrono>
+#include <functional>
 
+#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h"
+#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
 #include "utils/logs/logging.h"
 
 namespace softbus::message_bus::pipeline
 {
 
+struct EndpointFramingState
+{
+    softbus::core::memory::LockFreeQueue<PipelineContext> chunkQ;
+    std::atomic<bool>                                   running{true};
+    std::thread                                         th;
+    std::uint64_t                                       nextSeq{1};
+    QString                                             endpoint;
+    std::shared_ptr<softbus::core::memory::MemoryPool>             pool;
+    softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig modbusCfg;
+    std::shared_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> pass;
+    std::function<void(PipelineContext)> pushFramed;
+
+    explicit EndpointFramingState(std::size_t cap)
+        : chunkQ(cap)
+    {
+    }
+};
+
+namespace
+{
+
+void endpointFramerThread(std::shared_ptr<EndpointFramingState> st)
+{
+    using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer;
+    using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu;
+
+    auto modbus = std::make_unique<ModbusRtuFramer>(st->pool, st->modbusCfg, st->endpoint);
+
+    while (st->running.load(std::memory_order_acquire)) {
+        PipelineContext ch;
+        if (!st->chunkQ.pop(ch)) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            continue;
+        }
+        std::vector<PipelineContext> out;
+        if (protocolHintIsModbusRtu(ch.protocolHint)) {
+            modbus->feed(ch, out);
+        } else if (st->pass) {
+            st->pass->feed(ch, out);
+        }
+        for (auto& x : out) {
+            x.frameSeq = st->nextSeq++;
+            if (st->pushFramed) {
+                st->pushFramed(std::move(x));
+            }
+        }
+    }
+}
+
+} // namespace
+
 PipelineEngine::PipelineEngine(std::size_t queueCapacity)
-    : m_ingressQ(queueCapacity), m_egressQ(queueCapacity)
+    : m_ingressQ(queueCapacity)
+    , m_egressQ(queueCapacity)
 {
 }
 
@@ -17,12 +75,73 @@ PipelineEngine::~PipelineEngine()
     stop();
 }
 
+void PipelineEngine::setMemoryPool(std::shared_ptr<softbus::core::memory::MemoryPool> pool)
+{
+    m_pool = std::move(pool);
+}
+
+void PipelineEngine::setRuntimeConfig(PipelineRuntimeConfig cfg)
+{
+    m_rtConfig = cfg;
+}
+
+void PipelineEngine::loadRuntimeConfigFromJson(const QJsonObject& obj)
+{
+    if (obj.isEmpty()) {
+        LOG_ERROR() << "PipelineEngine: runtime config is empty";
+        // return;
+    }
+    m_rtConfig.parallelPipeline   = obj.value(QStringLiteral("parallelPipeline")).toBool(true);
+    m_rtConfig.parseWorkerCount   = std::max(1, obj.value(QStringLiteral("parseWorkerCount")).toInt(2));
+    m_rtConfig.orderedEgress      = obj.value(QStringLiteral("orderedEgress")).toBool(true);
+    m_rtConfig.framedQueueMax     = static_cast<std::size_t>(
+        std::max(64, obj.value(QStringLiteral("framedQueueMax")).toInt(4096)));
+    m_rtConfig.maxReorderDepth    = static_cast<std::size_t>(
+        std::max(16, obj.value(QStringLiteral("maxReorderDepth")).toInt(256)));
+    m_rtConfig.modbus.maxAdu              = static_cast<std::size_t>(
+        std::max(8, obj.value(QStringLiteral("modbusRtuMaxFrameBytes")).toInt(256)));
+    m_rtConfig.modbus.interFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0);
+}
+
 void PipelineEngine::start()
 {
     if (m_running.exchange(true)) {
         return;
     }
-    m_worker = std::thread([this]() { run(); });
+
+    if (!m_pool) {
+        LOG_ERROR() << "PipelineEngine::start: memory pool not set";
+        m_running = false;
+        return;
+    }
+
+    m_modbusFramer =
+        std::make_unique<softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer>(m_pool,
+                                                                                            m_rtConfig.modbus,
+                                                                                            QString{});
+    m_passthroughFramer =
+        std::make_shared<softbus::message_bus::pipeline::stages::framers::PassthroughFramer>(m_pool);
+
+    // 如果启用并行管道 则创建帧队列和解析线程 如果启用有序出口 则创建合并队列和合并线程
+    // 并行管道是用来处理多个端点的数据流的
+    // 有序出口是用来保证数据流的有序性的
+    // 帧队列是用来存储帧的 
+    // 解析线程是用来处理帧的
+    // 合并队列是用来存储合并的帧的
+    // 合并线程是用来处理合并的帧的
+    if (m_rtConfig.parallelPipeline) {
+        m_framedQ = std::make_unique<softbus::core::threading::MutexQueue<PipelineContext>>(
+            m_rtConfig.framedQueueMax);
+        for (int i = 0; i < m_rtConfig.parseWorkerCount; ++i) {
+            m_parseThreads.emplace_back([this]() { parseWorkerLoop(); });
+        }
+        if (m_rtConfig.orderedEgress) {
+            m_mergeQ        = std::make_unique<softbus::core::threading::MutexQueue<MergeItem>>(0);
+            m_mergerThread = std::thread([this]() { mergerLoop(); });
+        }
+    } else {
+        m_worker = std::thread([this]() { runSingleWorker(); });
+    }
 }
 
 void PipelineEngine::stop()
@@ -30,20 +149,77 @@ void PipelineEngine::stop()
     if (!m_running.exchange(false)) {
         return;
     }
+
+    {
+        std::lock_guard<std::mutex> lk(m_endpointMutex);
+        for (auto it = m_endpointFramers.constBegin(); it != m_endpointFramers.constEnd(); ++it) {
+            if (it.value()) {
+                it.value()->running.store(false, std::memory_order_release);
+            }
+        }
+        for (auto it = m_endpointFramers.constBegin(); it != m_endpointFramers.constEnd(); ++it) {
+            if (it.value() && it.value()->th.joinable()) {
+                it.value()->th.join();
+            }
+        }
+        m_endpointFramers.clear();
+    }
+
     if (m_worker.joinable()) {
         m_worker.join();
     }
-}
 
+    for (auto& t : m_parseThreads) {
+        if (t.joinable()) {
+            t.join();
+        }
+    }
+    m_parseThreads.clear();
+
+    if (m_mergerThread.joinable()) {
+        m_mergerThread.join();
+    }
+
+    m_framedQ.reset();
+    m_mergeQ.reset();
+    m_modbusFramer.reset();
+    m_passthroughFramer.reset();
+}
+// 功能 ：清空 ingress 队列
+// 参数 ：无
+// 返回值 ：无
+// 逻辑 ：从 ingress 队列中取出数据 直到队列为空
 void PipelineEngine::drain()
 {
     PipelineContext tmp;
     while (m_ingressQ.pop(tmp)) {
     }
 }
-
+// 功能 ：将数据推送到 ingress 队列
+// 参数 ：ctx 数据
+// 返回值 ：是否成功
+// 逻辑 ：如果并行管道启用 则根据端点确保帧解析器 然后推送到帧队列 如果并行管道未启用 则直接推送到 ingress 队列
 bool PipelineEngine::enqueueIngress(PipelineContext ctx)
 {
+    if (m_rtConfig.parallelPipeline) {
+        if (ctx.endpoint.isEmpty()) {
+            ++m_counters.drop;
+            return false;
+        }
+        ensureEndpointFramer(ctx.endpoint);
+        std::shared_ptr<EndpointFramingState> st;
+        {
+            std::lock_guard<std::mutex> lk(m_endpointMutex);
+            st = m_endpointFramers.value(ctx.endpoint);
+        }
+        if (!st || !st->chunkQ.push(std::move(ctx))) {
+            ++m_counters.drop;
+            return false;
+        }
+        ++m_counters.in;
+        return true;
+    }
+
     if (!m_ingressQ.push(std::move(ctx))) {
         ++m_counters.drop;
         return false;
@@ -51,41 +227,200 @@ bool PipelineEngine::enqueueIngress(PipelineContext ctx)
     ++m_counters.in;
     return true;
 }
-
+// 功能 ：将数据推送到 egress 队列
+// 参数 ：ctx 数据
+// 返回值 ：是否成功
+// 逻辑 ：将数据推送到 egress 队列
 bool PipelineEngine::enqueueEgress(PipelineContext ctx)
 {
     return m_egressQ.push(std::move(ctx));
 }
-
+// 功能 ：尝试从 egress 队列中取出数据
+// 参数 ：out 数据
+// 返回值 ：是否成功
+// 逻辑 ：从 egress 队列中取出数据 如果队列为空 则返回false 如果队列不为空 则返回true
 bool PipelineEngine::tryPopEgress(PipelineContext& out)
 {
     return m_egressQ.pop(out);
 }
-
+// 功能 ：设置插件管理器
+// 参数 ：pluginManager 插件管理器
+// 返回值 ：无
+// 逻辑 ：设置插件管理器
 void PipelineEngine::setPluginManager(std::shared_ptr<softbus::core::plugin_system::PluginManager> pluginManager)
 {
     m_pluginManager = std::move(pluginManager);
 }
 
-// 管道引擎运行线程 从 ingress 队列中取出数据 进行处理 然后推送到 egress 队列中
-void PipelineEngine::run()
+// 功能 ：确保端点帧解析器
+// 参数 ：endpoint 端点
+// 返回值 ：无
+// 逻辑 ：如果端点帧解析器不存在 则创建端点帧解析器
+void PipelineEngine::ensureEndpointFramer(const QString& endpoint)
 {
-    while (m_running.load()) {
+    std::lock_guard<std::mutex> lk(m_endpointMutex);
+    if (m_endpointFramers.contains(endpoint)) {
+        return;
+    }
+    const std::size_t qcap = m_ingressQ.capacity();
+    auto              st   = std::make_shared<EndpointFramingState>(qcap > 0 ? qcap : 4096);
+    st->endpoint           = endpoint;
+    st->pool               = m_pool;
+    st->modbusCfg          = m_rtConfig.modbus;
+    st->pass               = m_passthroughFramer;
+    st->pushFramed         = [this](PipelineContext c) {
+        if (!m_framedQ) {
+            return;
+        }
+        if (!m_framedQ->tryPush(std::move(c))) {
+            ++m_counters.drop;
+            LOG_WARNING() << "PipelineEngine: framed queue full drop";
+        }
+    };
+    st->th = std::thread(endpointFramerThread, st);
+    m_endpointFramers.insert(endpoint, std::move(st));
+}
+// 功能 ：分发帧解析器
+// 参数 ：chunkIn 数据
+// 返回值 ：无
+// 逻辑 ：根据协议提示 选择合适的帧解析器 进行帧解析
+void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector<PipelineContext>& completeFramesOut)
+{
+    using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu;
+    // 根据协议提示 选择合适的帧解析器 进行帧解析
+
+    if (protocolHintIsModbusRtu(chunkIn.protocolHint)) {
+        if (m_modbusFramer) {
+            m_modbusFramer->feed(chunkIn, completeFramesOut);
+
+        }
+    } else if (m_passthroughFramer) {
+        m_passthroughFramer->feed(chunkIn, completeFramesOut);
+
+    }
+}
+// 功能 ：分配帧序列号
+void PipelineEngine::assignFrameSequence(PipelineContext& ctx)
+{
+    std::lock_guard<std::mutex> lk(m_seqMutex);
+    ctx.frameSeq = ++m_nextFrameSeqByEndpoint[ctx.endpoint];
+}
+
+void PipelineEngine::runSingleWorker()
+{
+    // 单线程工作线程 从 ingress 队列中取出数据 进行处理 然后推送到 egress 队列中
+    while (m_running.load(std::memory_order_acquire)) {
         PipelineContext ctx;
         if (!m_ingressQ.pop(ctx)) {
             std::this_thread::sleep_for(std::chrono::milliseconds(1));
             continue;
         }
 
-        const bool framed = stage1Framer(ctx);
-        // 帧解析
-        const bool parsed = framed && stage2Parser(ctx);
-        // 过滤
-        const bool filtered = parsed && stage3Filter(ctx);
-        // 验证
-        const bool valid = filtered && stage4Validator(ctx);
-        // 如果验证失败 则记录错误
-        if (!valid) {
+        std::vector<PipelineContext> batch;
+        dispatchFramer(ctx, batch);
+
+        for (auto& c : batch) {
+            if (c.frameKind != FrameKind::CompleteFrame) {
+                continue;
+            }
+            // 
+            std::ostringstream oss;
+            oss << std::hex << std::setfill('0');
+            for (std::size_t i = 0; i < c.payload.block->size; ++i) {
+                oss << std::setw(2) << static_cast<int>(c.payload.block->data[i]);
+                if (i + 1 < c.payload.block->size) {
+                    oss << ' ';
+                }
+            }
+            const QString hexPreview = QString::fromStdString(oss.str());
+            LOG_INFO() << "the data is"<< hexPreview;
+            // 分配帧序列号
+            assignFrameSequence(c);
+            if (!runStages234(c)) {
+                ++m_counters.error;
+                continue;
+            }
+            if (!m_egressQ.push(std::move(c))) {
+                ++m_counters.drop;
+                continue;
+            }
+            ++m_counters.out;
+        }
+    }
+}
+
+void PipelineEngine::parseWorkerLoop()
+{
+    while (true) {
+        PipelineContext ctx;
+    
+        const bool      have = m_framedQ && m_framedQ->popWaitFor(ctx, m_running, std::chrono::milliseconds(50)); // 从帧队列中取出帧 , 如果帧队列为空 则等待50毫秒
+        if (!have) {
+            if (!m_running.load(std::memory_order_acquire)) {
+                break;
+            }
+            continue;
+        }
+
+
+        const bool ok = runStages234(ctx);
+        if (m_rtConfig.orderedEgress && m_mergeQ) {
+            MergeItem it;
+            it.endpoint    = ctx.endpoint;
+            it.seq         = ctx.frameSeq;
+            it.pipelineOk  = ok;
+            it.ctx         = std::move(ctx);
+            m_mergeQ->push(std::move(it));
+        } else {
+            if (!ok) {
+                ++m_counters.error;
+                continue;
+            }
+            if (!m_egressQ.push(std::move(ctx))) {
+                ++m_counters.drop;
+                continue;
+            }
+            ++m_counters.out;
+        }
+    }
+}
+
+void PipelineEngine::mergerLoop()
+{
+    while (true) {
+        MergeItem it;
+        const bool have = m_mergeQ && m_mergeQ->popWaitFor(it, m_running, std::chrono::milliseconds(50));
+        if (!have) {
+            if (!m_running.load(std::memory_order_acquire)) {
+                break;
+            }
+            continue;
+        }
+        ingestMergeItem(std::move(it));
+    }
+}
+// 功能 ：合并帧
+// 参数 ：item 合并项
+// 返回值 ：无
+// 逻辑 ：将合并项中的数据推送到 egress 队列中
+void PipelineEngine::ingestMergeItem(MergeItem item)
+{
+    std::lock_guard<std::mutex> lk(m_mergeStateMutex);
+    auto& st = m_mergeByEndpoint[item.endpoint];
+    if (m_rtConfig.maxReorderDepth > 0
+        && static_cast<std::size_t>(st.buffer.size()) >= m_rtConfig.maxReorderDepth) {
+        ++m_counters.mergeDrop;
+        LOG_WARNING() << "PipelineEngine: merge reorder depth exceeded endpoint=" << item.endpoint;
+        return;
+    }
+    st.buffer.insert(item.seq, std::make_pair(std::move(item.ctx), item.pipelineOk));
+
+    while (st.buffer.contains(st.nextExpected)) {
+        auto pr = st.buffer.take(st.nextExpected);
+        ++st.nextExpected;
+        const bool             pipelineOk = pr.second;
+        PipelineContext        ctx          = std::move(pr.first);
+        if (!pipelineOk) {
             ++m_counters.error;
             continue;
         }
@@ -97,17 +432,61 @@ void PipelineEngine::run()
     }
 }
 
-bool PipelineEngine::stage1Framer(PipelineContext& ctx)
+// 功能 ：运行阶段234
+// 参数 ：ctx 数据
+// 返回值 ：是否成功
+// 逻辑 ：如果帧类型不是完整帧 则返回false 如果阶段2解析失败 则返回false 如果阶段3过滤失败 则返回false 如果阶段4验证失败 则返回false 如果都成功 则返回true
+bool PipelineEngine::runStages234(PipelineContext& ctx)
 {
-    if (!ctx.payload.valid()) {
+    if (ctx.frameKind != FrameKind::CompleteFrame) {
         return false;
     }
-    if (ctx.payloadSize == 0) {
-        ctx.payloadSize = ctx.payload.length;
+    std::ostringstream oss;
+    oss << std::hex << std::setfill('0');
+    for (std::size_t i = 0; i < ctx.payload.block->size; ++i) {
+        oss << std::setw(2) << static_cast<int>(ctx.payload.block->data[i]);
+        if (i + 1 < ctx.payload.block->size) {
+            oss << ' ';
+        }
+    }
+    const QString hexPreview = QString::fromStdString(oss.str());
+    LOG_INFO() << "the data is"<< hexPreview;
+    LOG_INFO() << "the protocol hint is"<< ctx.protocolHint;
+    const QString directionText = (ctx.direction == BusDirection::Upstream)
+                                      ? QStringLiteral("Upstream")
+                                      : (ctx.direction == BusDirection::Downstream)
+                                            ? QStringLiteral("Downstream")
+                                            : QStringLiteral("Unknown");
+    LOG_INFO() << "the direction is" << directionText;
+    if (!stage2Parser(ctx)) {
+        return false;
+    }
+    QJsonObject parsed = ctx.parsed;
+    LOG_INFO() << "the parsed is"<< parsed;
+    
+    softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol = ctx.protocol;
+    if (const auto* modbusPdu = std::get_if<softbus::message_bus::pipeline::protocol::ModbusRtuPdu>(&protocol.pdu)) {
+        LOG_INFO() << "the protocol text is" << softbus::message_bus::pipeline::protocol::toText(*modbusPdu);
+    } else {
+        LOG_INFO() << "the protocol text is" << QStringLiteral("ProtocolEnvelopePdu::monostate");
+    }
+
+    m_count++;
+    LOG_INFO() << "the count of command is" << m_count;
+
+    if (!stage3Filter(ctx)) {
+        return false;
+    }
+    if (!stage4Validator(ctx)) {
+        return false;
     }
     return true;
 }
 
+// 功能 ：验证阶段4
+// 参数 ：ctx 数据
+// 返回值 ：是否成功
+// 逻辑 ：如果有效载荷无效 则返回false 如果都成功 则返回true
 bool PipelineEngine::stage4Validator(PipelineContext& ctx)
 {
     if (!ctx.payload.valid()) {
@@ -119,9 +498,16 @@ bool PipelineEngine::stage4Validator(PipelineContext& ctx)
     if (!ctx.timestamp.isValid()) {
         ctx.timestamp = QDateTime::currentDateTimeUtc();
     }
+    if (ctx.frameKind != FrameKind::CompleteFrame) {
+        return false;
+    }
     return true;
 }
 
+// 功能 ：运行阶段2解析器
+// 参数 ：ctx 数据
+// 返回值 ：是否成功
+// 逻辑 ：如果插件管理器为空 则返回true 如果插件管理器不为空 则选择合适的插件 创建会话 如果会话为空 则返回false 如果会话不为空 则返回true
 bool PipelineEngine::stage2Parser(PipelineContext& ctx)
 {
     if (!m_pluginManager) {
@@ -140,8 +526,8 @@ bool PipelineEngine::stage2Parser(PipelineContext& ctx)
 
 bool PipelineEngine::stage3Filter(PipelineContext& ctx)
 {
-    (void)ctx;
-    return true;
+    static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
+    return s_filter.apply(ctx);
 }
 
 } // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/PipelineEngine.h

@@ -1,21 +1,40 @@
 #pragma once
 // 定义管道引擎 包含队列、运行状态、工作线程、计数器、插件管理器
-// 队列是用来存储管道上下文的
-// 运行状态是用来控制引擎的运行和停止的
-// 工作线程是用来处理管道上下文的
-// 计数器是用来统计管道上下文的数量和错误数量的
-// 插件管理器是用来管理插件的
 #include <atomic>
 #include <memory>
 #include <thread>
+#include <utility>
+#include <vector>
+
+#include <QHash>
+#include <QJsonObject>
+#include <QString>
 
 #include "core/plugin_system/PluginManager.h"
 #include "core/memory/LockFreeQueue.h"
+#include "core/memory/MemoryPool.h"
+#include "core/threading/MutexQueue.h"
 #include "message_bus/pipeline/PipelineContext.h"
+#include "message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h"
+#include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h"
 
 namespace softbus::message_bus::pipeline
 {
 
+struct PipelineRuntimeConfig
+{
+    bool parallelPipeline{false};
+    int  parseWorkerCount{2};
+    bool orderedEgress{true};
+
+    std::size_t framedQueueMax{4096};
+    std::size_t maxReorderDepth{256};
+
+    softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig modbus;
+};
+
+struct EndpointFramingState;
+
 class PipelineEngine
 {
 public:
@@ -25,6 +44,7 @@ public:
         std::atomic<std::uint64_t> out{0};
         std::atomic<std::uint64_t> drop{0};
         std::atomic<std::uint64_t> error{0};
+        std::atomic<std::uint64_t> mergeDrop{0};
     };
 
     explicit PipelineEngine(std::size_t queueCapacity = 4096);
@@ -39,12 +59,38 @@ public:
 
     bool tryPopEgress(PipelineContext& out);
     void setPluginManager(std::shared_ptr<softbus::core::plugin_system::PluginManager> pluginManager);
+    void setMemoryPool(std::shared_ptr<softbus::core::memory::MemoryPool> pool);
+    void setRuntimeConfig(PipelineRuntimeConfig cfg);
+    void loadRuntimeConfigFromJson(const QJsonObject& obj);
 
     const Counters& counters() const { return m_counters; }
 
 private:
-    void run();
-    bool stage1Framer(PipelineContext& ctx);
+    struct MergeItem
+    {
+        QString          endpoint;
+        std::uint64_t    seq{0};
+        PipelineContext ctx;
+        bool             pipelineOk{false};
+    };
+
+    struct MergeState
+    {
+        std::uint64_t                                            nextExpected{1};
+        QHash<std::uint64_t, std::pair<PipelineContext, bool>> buffer;
+    };
+
+    void runSingleWorker();
+    void parseWorkerLoop();
+    void mergerLoop();
+
+    void ensureEndpointFramer(const QString& endpoint);
+
+    void dispatchFramer(PipelineContext& chunkIn, std::vector<PipelineContext>& completeFramesOut);
+    void assignFrameSequence(PipelineContext& ctx);
+    bool runStages234(PipelineContext& ctx);
+    void ingestMergeItem(MergeItem item);
+
     bool stage2Parser(PipelineContext& ctx);
     bool stage3Filter(PipelineContext& ctx);
     bool stage4Validator(PipelineContext& ctx);
@@ -53,9 +99,30 @@ private:
     softbus::core::memory::LockFreeQueue<PipelineContext> m_ingressQ;
     softbus::core::memory::LockFreeQueue<PipelineContext> m_egressQ;
     std::atomic<bool> m_running{false};
-    std::thread m_worker;
-    Counters m_counters;
+    std::thread       m_worker;
+    Counters          m_counters;
     std::shared_ptr<softbus::core::plugin_system::PluginManager> m_pluginManager;
+    std::shared_ptr<softbus::core::memory::MemoryPool>           m_pool;
+
+    PipelineRuntimeConfig m_rtConfig;
+
+    std::unique_ptr<softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer> m_modbusFramer;
+    std::shared_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> m_passthroughFramer;
+
+    std::mutex                    m_seqMutex;
+    QHash<QString, std::uint64_t> m_nextFrameSeqByEndpoint;
+
+    std::unique_ptr<softbus::core::threading::MutexQueue<PipelineContext>> m_framedQ;
+    std::unique_ptr<softbus::core::threading::MutexQueue<MergeItem>>       m_mergeQ;
+    std::vector<std::thread>                                                m_parseThreads;
+    std::thread                                                             m_mergerThread;
+
+    std::mutex                                              m_endpointMutex;
+    QHash<QString, std::shared_ptr<EndpointFramingState>> m_endpointFramers;
+
+    std::mutex                 m_mergeStateMutex;
+    QHash<QString, MergeState> m_mergeByEndpoint;
+    int m_count{0}; // 计数器
 };
 
 } // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/stages/1_framers/DefaultFramer.cpp

@@ -1,8 +0,0 @@
-#include "message_bus/pipeline/stages/1_framers/IFramer.h"
-
-namespace softbus::message_bus::pipeline::stages::framers
-{
-
-// Placeholder TU for stage1 concrete framer implementations.
-
-} // namespace softbus::message_bus::pipeline::stages::framers

src/message_bus/pipeline/stages/1_framers/IFramer.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <vector>
+
 #include "message_bus/pipeline/PipelineContext.h"
 
 namespace softbus::message_bus::pipeline::stages::framers
@@ -9,7 +11,10 @@ class IFramer
 {
 public:
     virtual ~IFramer() = default;
-    virtual bool frame(softbus::message_bus::pipeline::PipelineContext& ctx) = 0;
+
+    /// Consumes `chunkIn` (StreamChunk 语义)；仅在 `completeFramesOut` 中追加 CompleteFrame。
+    virtual void feed(softbus::message_bus::pipeline::PipelineContext&           chunkIn,
+                      std::vector<softbus::message_bus::pipeline::PipelineContext>& completeFramesOut) = 0;
 };
 
 } // namespace softbus::message_bus::pipeline::stages::framers

src/message_bus/pipeline/stages/2_parsers/IParserStage.h

@@ -1,5 +1,5 @@
 #pragma once
-
+// 定义解析器阶段接口
 #include "message_bus/pipeline/PipelineContext.h"
 
 namespace softbus::message_bus::pipeline::stages::parsers