diff --git a/CMakeLists.txt b/CMakeLists.txt index 4683010..66442e3 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -150,8 +150,7 @@ qt_add_executable(softbus_daemon src/core/metadata/ProfileRegistry.cpp # message bus pipeline - src/message_bus/pipeline/PayloadRef.h - src/message_bus/pipeline/PipelineContext.h + src/core/models/PayloadRef.h src/message_bus/pipeline/DynamicRuleRegistry.h src/message_bus/pipeline/DynamicRuleRegistry.cpp src/message_bus/pipeline/PipelineEngine.h diff --git a/bin/softbus_daemon b/bin/softbus_daemon index 17d8ea2..6137844 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/docs/数据流转.md b/docs/数据流转.md index 2cfb21d..5a1575c 100644 --- a/docs/数据流转.md +++ b/docs/数据流转.md @@ -70,7 +70,7 @@ 输出:`frameKind = CompleteFrame` 的帧批次。 -### 3.3 Stage 2/3/4:解析 -> 过滤 -> 校验 +### 3.3 Stage 2/3/4:解析 -> 过滤 -> 校验(当前返回语义) `runStages234(...)` 核心流程: @@ -83,10 +83,13 @@ 3. **Stage4 Validator** - 最低校验当前为 `messageId` 非空。 4. **Transform/Enrich** - - 应用动态规则(scale/offset/unit)。 + - 仅对通过 filter+validator 的 `DOMMessage` 应用动态规则(scale/offset/unit)。 - 写入 trace,补充富化字段。 -说明:当前 `runStages234` 只返回布尔结果,`EnrichedMessage` 由内部生成,后续可接持久化/发布环节。 +说明(与代码一致): +- `runStages234` 在 `stage2Parser` 返回空时才返回 `false`。 +- 只要 `stage2Parser` 有消息,哪怕全部在 Stage3/Stage4 被跳过,当前实现仍返回 `true`。 +- `EnrichedMessage` 在内部生成,但尚未进入独立发布/持久化通道。 ### 3.4 出口与有序保证 @@ -101,12 +104,12 @@ ## 4. 错误与背压点 -主要失败点及表现: +主要失败点及表现(按当前实现): 1. **内存池分配失败**:`SerialDevice::onRead` 直接 `reportError(memory pool exhausted)`。 2. **入口队列满**:`enqueueIngress` 返回 false,计入 drop。 -3. **分帧/解析插件缺失或失败**:对应帧在阶段中被丢弃。 -4. **过滤/校验失败**:消息跳过,不进入下游。 +3. **分帧/解析/映射插件缺失或失败**:`stage2Parser` 返回空,`runStages234=false`,该帧不进入 egress。 +4. **过滤/校验失败**:对应 `DOMMessage` 被跳过(不做 enrich),但该 `PipelineContext` 仍可能继续进入 egress(只要 Stage2 非空)。 5. **重排深度超限**:`mergeDrop` 增加并告警。 --- @@ -125,16 +128,19 @@ SerialQtDriver(read callback) -> [Stage4] validate -> transform/enrich -> egressQ (有序模式下先 merge 再 egress) + -> DeviceBusManager::processEgressLoop + -> EgressRouter::send + -> SerialDevice::writePayload ``` --- -## 6. 下行(现状) +## 6. egress 到设备(现状) -- `RawBusMessage.direction` 已支持 `Downstream` 语义。 -- `SerialDevice` 已提供 `writePayload(...)` 写入能力。 -- 当前代码中 egress 路由到物理设备的绑定仍预留(`SerialDeviceManager` 里相关逻辑已注释),后续可补齐: - - egress 消息 -> `EgressRouter` -> endpoint 设备 -> `writePayload`。 +- `DeviceBusManager` 已启动 `processEgressLoop`,持续从 `PipelineEngine::tryPopEgress` 取数据并调用 `EgressRouter::send`。 +- `EgressRouter` 已实现 `endpoint -> SerialDevice` 查找和 `writePayload(...)` 发送。 +- 当前关键缺口:`SerialDeviceManager::initializeSerialDeviceWithDeps` 中 `bindSerialDevice(...)` 仍被注释,导致路由表通常为空,`send` 多数返回失败。 +- 因此当前链路是“egress 框架已存在,但默认没有绑定到串口实例”。 --- diff --git a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp index d9b4816..51465e2 100644 --- a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp +++ b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp @@ -25,10 +25,9 @@ public: m_buf.resize(m_cfg.maxAdu); } - void feedChunk(softbus::message_bus::pipeline::PipelineContext& in, - std::vector& out) override + void feedChunk(softbus::core::models::RawBusMessage& in, + std::vector& out) override { - using softbus::message_bus::pipeline::FrameKind; if (!m_pool || !in.payload.valid()) { return; } @@ -77,7 +76,7 @@ public: std::memcpy(blk->data, m_buf.data(), *pred); blk->size = *pred; - softbus::message_bus::pipeline::PipelineContext framed; + softbus::core::models::RawBusMessage framed; framed.deviceId = in.deviceId; framed.stableKey = in.stableKey; framed.endpoint = in.endpoint; @@ -85,7 +84,6 @@ public: framed.protocolHint = in.protocolHint; framed.traceId = in.traceId; framed.direction = in.direction; - framed.frameKind = FrameKind::CompleteFrame; framed.payload.block = std::move(blk); framed.payload.offset = 0; framed.payload.length = *pred; @@ -101,7 +99,6 @@ public: in.payload = {}; in.payloadSize = 0; - in.frameKind = FrameKind::StreamChunk; } void reset() override { m_len = 0; } diff --git a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp index 5733338..eae4f86 100644 --- a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp +++ b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp @@ -12,7 +12,7 @@ class ModbusRtuMapperSession final : public softbus::core::plugin_system::IProto { public: std::vector map( - const softbus::message_bus::pipeline::PipelineContext& rawContext, + const softbus::core::models::RawBusMessage& rawContext, const softbus::message_bus::pipeline::protocol::ProtocolEnvelope& envelope) override { std::vector out; diff --git a/src/core/CoreService.cpp b/src/core/CoreService.cpp index 57b4a83..5d98dc8 100644 --- a/src/core/CoreService.cpp +++ b/src/core/CoreService.cpp @@ -10,8 +10,8 @@ #include "utils/logs/logging.h" #include "device_bus/DeviceBus.h" -#include "message_bus/pipeline/PayloadRef.h" -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/PayloadRef.h" +#include "core/models/RawBusMessage.h" CoreService* CoreService::instance() { @@ -38,8 +38,8 @@ bool CoreService::initialize() } LOG_INFO() << "CoreService initializing"; - qRegisterMetaType("softbus::message_bus::pipeline::PayloadRef"); - qRegisterMetaType("softbus::message_bus::pipeline::PipelineContext"); + qRegisterMetaType("softbus::core::models::PayloadRef"); + qRegisterMetaType("softbus::core::models::RawBusMessage"); // 1. 解析配置路径(放在用户 home 下的相对固定目录:~/softbus/config) const QString configPath = diff --git a/src/message_bus/pipeline/PayloadRef.h b/src/core/models/PayloadRef.h similarity index 69% rename from src/message_bus/pipeline/PayloadRef.h rename to src/core/models/PayloadRef.h index 1203e70..0bee0ad 100644 --- a/src/message_bus/pipeline/PayloadRef.h +++ b/src/core/models/PayloadRef.h @@ -1,14 +1,13 @@ #pragma once -// 功能 :定义有效载荷引用 包含数据块、偏移量、长度 + #include #include -#include #include #include "core/memory/MemoryPool.h" -namespace softbus::message_bus::pipeline +namespace softbus::core::models { struct PayloadRef @@ -31,6 +30,6 @@ struct PayloadRef } }; -} // namespace softbus::message_bus::pipeline +} // namespace softbus::core::models -Q_DECLARE_METATYPE(softbus::message_bus::pipeline::PayloadRef) +Q_DECLARE_METATYPE(softbus::core::models::PayloadRef) diff --git a/src/core/models/RawBusMessage.h b/src/core/models/RawBusMessage.h index 3b38e70..a3c9d04 100644 --- a/src/core/models/RawBusMessage.h +++ b/src/core/models/RawBusMessage.h @@ -6,7 +6,7 @@ #include #include -#include "message_bus/pipeline/PayloadRef.h" +#include "core/models/PayloadRef.h" namespace softbus::core::models { @@ -27,7 +27,7 @@ struct RawBusMessage QString protocolHint; QString traceId; - softbus::message_bus::pipeline::PayloadRef payload; + softbus::core::models::PayloadRef payload; std::size_t payloadSize{0}; BusDirection direction{BusDirection::Unknown}; diff --git a/src/core/plugin_system/IProtocolFramerPlugin.h b/src/core/plugin_system/IProtocolFramerPlugin.h index bcf8c8b..a32f049 100644 --- a/src/core/plugin_system/IProtocolFramerPlugin.h +++ b/src/core/plugin_system/IProtocolFramerPlugin.h @@ -6,8 +6,8 @@ #include #include +#include "core/models/RawBusMessage.h" #include "core/memory/MemoryPool.h" -#include "message_bus/pipeline/PipelineContext.h" namespace softbus::core::plugin_system { @@ -17,8 +17,8 @@ class IProtocolFramerSession public: virtual ~IProtocolFramerSession() = default; - virtual void feedChunk(softbus::message_bus::pipeline::PipelineContext& in, - std::vector& out) = 0; + virtual void feedChunk(softbus::core::models::RawBusMessage& in, + std::vector& out) = 0; virtual void reset() = 0; virtual void shutdown() = 0; }; diff --git a/src/core/plugin_system/IProtocolMapperPlugin.h b/src/core/plugin_system/IProtocolMapperPlugin.h index 4c417c3..45aecdc 100644 --- a/src/core/plugin_system/IProtocolMapperPlugin.h +++ b/src/core/plugin_system/IProtocolMapperPlugin.h @@ -6,7 +6,7 @@ #include #include "core/models/DOMMessage.h" -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/RawBusMessage.h" #include "message_bus/pipeline/protocol/ProtocolEnvelope.h" namespace softbus::core::plugin_system @@ -17,7 +17,7 @@ class IProtocolMapperSession public: virtual ~IProtocolMapperSession() = default; virtual std::vector map( - const softbus::message_bus::pipeline::PipelineContext& rawContext, + const softbus::core::models::RawBusMessage& rawContext, const softbus::message_bus::pipeline::protocol::ProtocolEnvelope& envelope) = 0; virtual void reset() = 0; }; diff --git a/src/device_bus/manager/DeviceBusManager.cpp b/src/device_bus/manager/DeviceBusManager.cpp index b18d71e..6a6f46f 100644 --- a/src/device_bus/manager/DeviceBusManager.cpp +++ b/src/device_bus/manager/DeviceBusManager.cpp @@ -241,7 +241,7 @@ void DeviceBusManager::processEgressLoop() continue; } - softbus::message_bus::pipeline::PipelineContext out; + softbus::core::models::RawBusMessage out; if (!m_pipelineEngine->tryPopEgress(out)) { std::this_thread::sleep_for(std::chrono::milliseconds(1)); continue; diff --git a/src/message_bus/egress/EgressRouter.cpp b/src/message_bus/egress/EgressRouter.cpp index 9e3c6fb..e0fae67 100644 --- a/src/message_bus/egress/EgressRouter.cpp +++ b/src/message_bus/egress/EgressRouter.cpp @@ -17,7 +17,7 @@ void EgressRouter::unbindSerialDevice(const QString& endpoint) m_serialByEndpoint.remove(endpoint); } -bool EgressRouter::send(const softbus::message_bus::pipeline::PipelineContext& ctx) +bool EgressRouter::send(const softbus::core::models::RawBusMessage& ctx) { auto it = m_serialByEndpoint.constFind(ctx.endpoint); if (it == m_serialByEndpoint.constEnd() || !it.value() || !ctx.payload.valid()) { diff --git a/src/message_bus/egress/EgressRouter.h b/src/message_bus/egress/EgressRouter.h index 4168eb5..6a44e03 100644 --- a/src/message_bus/egress/EgressRouter.h +++ b/src/message_bus/egress/EgressRouter.h @@ -21,7 +21,7 @@ public: void bindSerialDevice(const QString& endpoint, const QSharedPointer& device); void unbindSerialDevice(const QString& endpoint); - bool send(const softbus::message_bus::pipeline::PipelineContext& ctx) override; + bool send(const softbus::core::models::RawBusMessage& ctx) override; private: QHash> m_serialByEndpoint; diff --git a/src/message_bus/egress/IEgressPort.h b/src/message_bus/egress/IEgressPort.h index cc1cb52..5bcb5fc 100644 --- a/src/message_bus/egress/IEgressPort.h +++ b/src/message_bus/egress/IEgressPort.h @@ -1,6 +1,6 @@ #pragma once -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/RawBusMessage.h" namespace softbus::message_bus::egress { @@ -9,7 +9,7 @@ class IEgressPort { public: virtual ~IEgressPort() = default; - virtual bool send(const softbus::message_bus::pipeline::PipelineContext& ctx) = 0; + virtual bool send(const softbus::core::models::RawBusMessage& ctx) = 0; }; } // namespace softbus::message_bus::egress diff --git a/src/message_bus/ingress/DriverIngressAdapter.cpp b/src/message_bus/ingress/DriverIngressAdapter.cpp index cc12aff..1d052bd 100644 --- a/src/message_bus/ingress/DriverIngressAdapter.cpp +++ b/src/message_bus/ingress/DriverIngressAdapter.cpp @@ -17,9 +17,7 @@ bool DriverIngressAdapter::push(softbus::core::models::RawBusMessage ctx) if (!m_engine) { return false; } - softbus::message_bus::pipeline::PipelineContext pipelineCtx; - pipelineCtx.raw() = std::move(ctx); - return m_engine->enqueueIngress(std::move(pipelineCtx)); + return m_engine->enqueueIngress(std::move(ctx)); } void DriverIngressAdapter::reportError(const QString& endpoint, const QString& error) diff --git a/src/message_bus/pipeline/PipelineContext.h b/src/message_bus/pipeline/PipelineContext.h deleted file mode 100644 index 0024ae8..0000000 --- a/src/message_bus/pipeline/PipelineContext.h +++ /dev/null @@ -1,29 +0,0 @@ -#pragma once - -#include - -#include -#include "core/models/RawBusMessage.h" - -namespace softbus::message_bus::pipeline -{ - -enum class FrameKind -{ - StreamChunk, - CompleteFrame, -}; - -using BusDirection = softbus::core::models::BusDirection; - -struct PipelineContext : public softbus::core::models::RawBusMessage -{ - FrameKind frameKind{FrameKind::StreamChunk}; - std::uint64_t frameSeq{0}; - softbus::core::models::RawBusMessage& raw() { return *this; } - const softbus::core::models::RawBusMessage& raw() const { return *this; } -}; - -} // namespace softbus::message_bus::pipeline - -Q_DECLARE_METATYPE(softbus::message_bus::pipeline::PipelineContext) diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index 717c207..a85b23d 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -13,14 +13,14 @@ namespace softbus::message_bus::pipeline struct EndpointFramingState { - softbus::core::memory::LockFreeQueue chunkQ; + softbus::core::memory::LockFreeQueue chunkQ; std::atomic running{true}; std::thread th; std::uint64_t nextSeq{1}; QString endpoint; std::shared_ptr pool; - std::function&)> feed; - std::function pushFramed; + std::function&)> feed; + std::function pushFramed; explicit EndpointFramingState(std::size_t cap) : chunkQ(cap) @@ -34,19 +34,19 @@ namespace void endpointFramerThread(std::shared_ptr st) { while (st->running.load(std::memory_order_acquire)) { - PipelineContext ch; + softbus::core::models::RawBusMessage ch; if (!st->chunkQ.pop(ch)) { std::this_thread::sleep_for(std::chrono::milliseconds(1)); continue; } - std::vector out; + std::vector out; if (st->feed) { st->feed(ch, out); } for (auto& x : out) { - x.frameSeq = st->nextSeq++; + const std::uint64_t seq = st->nextSeq++; if (st->pushFramed) { - st->pushFramed(std::move(x)); + st->pushFramed(st->endpoint, seq, std::move(x)); } } } @@ -113,7 +113,7 @@ void PipelineEngine::start() // 合并队列是用来存储合并的帧的 // 合并线程是用来处理合并的帧的 if (m_rtConfig.parallelPipeline) { - m_framedQ = std::make_unique>(m_rtConfig.framedQueueMax); + m_framedQ = std::make_unique>(m_rtConfig.framedQueueMax); for (int i = 0; i < m_rtConfig.parseWorkerCount; ++i) { // 创建解析线程 解析线程从帧队列中取出帧 进行处理 // emplace_back 是用来创建解析线程的 参数是 lambda 表达式 @@ -187,7 +187,7 @@ void PipelineEngine::stop() // 逻辑 :从 ingress 队列中取出数据 直到队列为空 void PipelineEngine::drain() { - PipelineContext tmp; + softbus::core::models::RawBusMessage tmp; while (m_ingressQ.pop(tmp)) { } } @@ -195,7 +195,7 @@ void PipelineEngine::drain() // 参数 :ctx 数据 // 返回值 :是否成功 // 逻辑 :如果并行管道启用 则根据端点确保帧解析器 然后推送到帧队列 如果并行管道未启用 则直接推送到 ingress 队列 -bool PipelineEngine::enqueueIngress(PipelineContext ctx) +bool PipelineEngine::enqueueIngress(softbus::core::models::RawBusMessage ctx) { if (m_rtConfig.parallelPipeline) { if (ctx.endpoint.isEmpty()) { @@ -227,7 +227,7 @@ bool PipelineEngine::enqueueIngress(PipelineContext ctx) // 参数 :ctx 数据 // 返回值 :是否成功 // 逻辑 :将数据推送到 egress 队列 -bool PipelineEngine::enqueueEgress(PipelineContext ctx) +bool PipelineEngine::enqueueEgress(softbus::core::models::RawBusMessage ctx) { return m_egressQ.push(std::move(ctx)); } @@ -235,7 +235,7 @@ bool PipelineEngine::enqueueEgress(PipelineContext ctx) // 参数 :out 数据 // 返回值 :是否成功 // 逻辑 :从 egress 队列中取出数据 如果队列为空 则返回false 如果队列不为空 则返回true -bool PipelineEngine::tryPopEgress(PipelineContext& out) +bool PipelineEngine::tryPopEgress(softbus::core::models::RawBusMessage& out) { return m_egressQ.pop(out); } @@ -262,14 +262,19 @@ void PipelineEngine::ensureEndpointFramer(const QString& endpoint) auto st = std::make_shared(qcap > 0 ? qcap : 4096); st->endpoint = endpoint; st->pool = m_pool; - st->feed = [this](PipelineContext& in, std::vector& out) { + st->feed = [this](softbus::core::models::RawBusMessage& in, std::vector& out) { this->dispatchFramer(in, out); }; - st->pushFramed = [this](PipelineContext c) { + st->pushFramed = [this](QString endpoint, std::uint64_t seq, softbus::core::models::RawBusMessage c) { if (!m_framedQ) { return; } - if (!m_framedQ->tryPush(std::move(c))) { + MergeItem item; + item.endpoint = std::move(endpoint); + item.seq = seq; + item.ctx = std::move(c); + item.pipelineOk = true; + if (!m_framedQ->tryPush(std::move(item))) { ++m_counters.drop; LOG_WARNING() << "PipelineEngine: framed queue full drop"; } @@ -281,7 +286,8 @@ void PipelineEngine::ensureEndpointFramer(const QString& endpoint) // 参数 :chunkIn 数据 // 返回值 :无 // 逻辑 :根据协议提示 选择合适的帧解析器 进行帧解析 -void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector& completeFramesOut) +void PipelineEngine::dispatchFramer( + softbus::core::models::RawBusMessage& chunkIn, std::vector& completeFramesOut) { if (m_pluginManager) { auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocolHint); @@ -307,29 +313,26 @@ void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector lk(m_seqMutex); - ctx.frameSeq = ++m_nextFrameSeqByEndpoint[ctx.endpoint]; + return ++m_nextFrameSeqByEndpoint[endpoint]; } void PipelineEngine::runSingleWorker() { // 单线程工作线程 从 ingress 队列中取出数据 进行处理 然后推送到 egress 队列中 while (m_running.load(std::memory_order_acquire)) { - PipelineContext ctx; + softbus::core::models::RawBusMessage ctx; if (!m_ingressQ.pop(ctx)) { std::this_thread::sleep_for(std::chrono::milliseconds(1)); continue; } - std::vector batch; + std::vector batch; dispatchFramer(ctx, batch); for (auto& c : batch) { - if (c.frameKind != FrameKind::CompleteFrame) { - continue; - } // std::ostringstream oss; oss << std::hex << std::setfill('0'); @@ -342,8 +345,8 @@ void PipelineEngine::runSingleWorker() const QString hexPreview = QString::fromStdString(oss.str()); LOG_INFO() << "the data is"<< hexPreview; // 分配帧序列号 - assignFrameSequence(c); - if (!runStages234(c)) { + const bool ok = runStages234(c); + if (!ok) { ++m_counters.error; continue; } @@ -359,9 +362,8 @@ void PipelineEngine::runSingleWorker() void PipelineEngine::parseWorkerLoop() { while (true) { - PipelineContext ctx; - - const bool have = m_framedQ && m_framedQ->popWaitFor(ctx, m_running, std::chrono::milliseconds(50)); // 从帧队列中取出帧 , 如果帧队列为空 则等待50毫秒 + MergeItem item; + const bool have = m_framedQ && m_framedQ->popWaitFor(item, m_running, std::chrono::milliseconds(50)); if (!have) { if (!m_running.load(std::memory_order_acquire)) { break; @@ -369,21 +371,16 @@ void PipelineEngine::parseWorkerLoop() continue; } - - const bool ok = runStages234(ctx); + const bool ok = runStages234(item.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)); + item.pipelineOk = ok; + m_mergeQ->push(std::move(item)); } else { if (!ok) { ++m_counters.error; continue; } - if (!m_egressQ.push(std::move(ctx))) { + if (!m_egressQ.push(std::move(item.ctx))) { ++m_counters.drop; continue; } @@ -426,7 +423,7 @@ void PipelineEngine::ingestMergeItem(MergeItem item) auto pr = st.buffer.take(st.nextExpected); ++st.nextExpected; const bool pipelineOk = pr.second; - PipelineContext ctx = std::move(pr.first); + softbus::core::models::RawBusMessage ctx = std::move(pr.first); if (!pipelineOk) { ++m_counters.error; continue; @@ -443,11 +440,8 @@ void PipelineEngine::ingestMergeItem(MergeItem item) // 参数 :ctx 数据 // 返回值 :是否成功 // 逻辑 :如果帧类型不是完整帧 则返回false 如果阶段2解析失败 则返回false 如果阶段3过滤失败 则返回false 如果阶段4验证失败 则返回false 如果都成功 则返回true -bool PipelineEngine::runStages234(PipelineContext& ctx) +bool PipelineEngine::runStages234(const softbus::core::models::RawBusMessage& ctx) { - if (ctx.frameKind != FrameKind::CompleteFrame) { - return false; - } auto messages = stage2Parser(ctx); LOG_INFO() << "PipelineEngine: stage2Parser messages size=" << messages.size(); if (messages.empty()) { @@ -478,7 +472,7 @@ bool PipelineEngine::stage4Validator(const softbus::core::models::DOMMessage& me // 参数 :ctx 数据 // 返回值 :是否成功 // 逻辑 :如果插件管理器为空 则返回true 如果插件管理器不为空 则选择合适的插件 创建会话 如果会话为空 则返回false 如果会话不为空 则返回true -std::vector PipelineEngine::stage2Parser(const PipelineContext& ctx) +std::vector PipelineEngine::stage2Parser(const softbus::core::models::RawBusMessage& ctx) { if (!m_pluginManager) { return {}; @@ -505,7 +499,7 @@ std::vector PipelineEngine::stage2Parser(cons parserSession = it.value(); } softbus::message_bus::pipeline::protocol::ProtocolEnvelope envelope; // 协议封装 - if (!parserSession->parse(ctx.raw(), envelope)) { + if (!parserSession->parse(ctx, envelope)) { return {}; } @@ -540,7 +534,7 @@ bool PipelineEngine::stage3Filter(softbus::core::models::DOMMessage& message) } softbus::core::models::EnrichedMessage PipelineEngine::applyTransformAndEnrich( - const PipelineContext& rawCtx, softbus::core::models::DOMMessage& message) const + const softbus::core::models::RawBusMessage& rawCtx, softbus::core::models::DOMMessage& message) const { auto rules = softbus::message_bus::pipeline::DynamicRuleRegistry::instance().snapshotTransformRules( rawCtx.endpoint, rawCtx.deviceId, rawCtx.stableKey); diff --git a/src/message_bus/pipeline/PipelineEngine.h b/src/message_bus/pipeline/PipelineEngine.h index dcaf9cf..ce8861a 100644 --- a/src/message_bus/pipeline/PipelineEngine.h +++ b/src/message_bus/pipeline/PipelineEngine.h @@ -15,11 +15,11 @@ #include "core/plugin_system/IProtocolMapperPlugin.h" #include "core/models/DOMMessage.h" #include "core/models/EnrichedMessage.h" +#include "core/models/RawBusMessage.h" #include "core/memory/LockFreeQueue.h" #include "core/memory/MemoryPool.h" #include "core/threading/MutexQueue.h" #include "message_bus/pipeline/DynamicRuleRegistry.h" -#include "message_bus/pipeline/PipelineContext.h" #include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h" namespace softbus::message_bus::pipeline @@ -56,10 +56,10 @@ public: void stop(); void drain(); - bool enqueueIngress(PipelineContext ctx); - bool enqueueEgress(PipelineContext ctx); + bool enqueueIngress(softbus::core::models::RawBusMessage ctx); + bool enqueueEgress(softbus::core::models::RawBusMessage ctx); - bool tryPopEgress(PipelineContext& out); + bool tryPopEgress(softbus::core::models::RawBusMessage& out); void setPluginManager(std::shared_ptr pluginManager); void setMemoryPool(std::shared_ptr pool); void setRuntimeConfig(PipelineRuntimeConfig cfg); @@ -72,14 +72,14 @@ private: { QString endpoint; std::uint64_t seq{0}; - PipelineContext ctx; + softbus::core::models::RawBusMessage ctx; bool pipelineOk{false}; }; struct MergeState { std::uint64_t nextExpected{1}; - QHash> buffer; + QHash> buffer; }; void runSingleWorker(); @@ -88,20 +88,21 @@ private: void ensureEndpointFramer(const QString& endpoint); - void dispatchFramer(PipelineContext& chunkIn, std::vector& completeFramesOut); - void assignFrameSequence(PipelineContext& ctx); - bool runStages234(PipelineContext& ctx); + void dispatchFramer(softbus::core::models::RawBusMessage& chunkIn, + std::vector& completeFramesOut); + std::uint64_t assignFrameSequence(const QString& endpoint); + bool runStages234(const softbus::core::models::RawBusMessage& ctx); void ingestMergeItem(MergeItem item); - std::vector stage2Parser(const PipelineContext& ctx); + std::vector stage2Parser(const softbus::core::models::RawBusMessage& ctx); bool stage3Filter(softbus::core::models::DOMMessage& message); bool stage4Validator(const softbus::core::models::DOMMessage& message) const; softbus::core::models::EnrichedMessage applyTransformAndEnrich( - const PipelineContext& rawCtx, softbus::core::models::DOMMessage& message) const; + const softbus::core::models::RawBusMessage& rawCtx, softbus::core::models::DOMMessage& message) const; private: - softbus::core::memory::LockFreeQueue m_ingressQ; - softbus::core::memory::LockFreeQueue m_egressQ; + softbus::core::memory::LockFreeQueue m_ingressQ; + softbus::core::memory::LockFreeQueue m_egressQ; std::atomic m_running{false}; std::thread m_worker; Counters m_counters; @@ -121,7 +122,7 @@ private: std::mutex m_seqMutex; QHash m_nextFrameSeqByEndpoint; - std::unique_ptr> m_framedQ; + std::unique_ptr> m_framedQ; std::unique_ptr> m_mergeQ; std::vector m_parseThreads; std::thread m_mergerThread; diff --git a/src/message_bus/pipeline/stages/1_framers/IFramer.h b/src/message_bus/pipeline/stages/1_framers/IFramer.h index 34209f7..e950790 100644 --- a/src/message_bus/pipeline/stages/1_framers/IFramer.h +++ b/src/message_bus/pipeline/stages/1_framers/IFramer.h @@ -2,7 +2,7 @@ #include -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/RawBusMessage.h" namespace softbus::message_bus::pipeline::stages::framers { @@ -13,8 +13,8 @@ public: virtual ~IFramer() = default; /// Consumes `chunkIn` (StreamChunk 语义);仅在 `completeFramesOut` 中追加 CompleteFrame。 - virtual void feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, - std::vector& completeFramesOut) = 0; + virtual void feed(softbus::core::models::RawBusMessage& chunkIn, + std::vector& completeFramesOut) = 0; }; } // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp index 614d24b..f82a689 100644 --- a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp +++ b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp @@ -10,8 +10,8 @@ PassthroughFramer::PassthroughFramer(std::shared_ptr& completeFramesOut) +void PassthroughFramer::feed(softbus::core::models::RawBusMessage& chunkIn, + std::vector& completeFramesOut) { if (!m_pool || !chunkIn.payload.valid()) { return; @@ -31,7 +31,7 @@ void PassthroughFramer::feed(softbus::message_bus::pipeline::PipelineContext& std::memcpy(blk->data, src, n); blk->size = n; - softbus::message_bus::pipeline::PipelineContext out; + softbus::core::models::RawBusMessage out; out.deviceId = chunkIn.deviceId; out.stableKey = chunkIn.stableKey; out.endpoint = chunkIn.endpoint; @@ -39,7 +39,6 @@ void PassthroughFramer::feed(softbus::message_bus::pipeline::PipelineContext& out.protocolHint = chunkIn.protocolHint; out.traceId = chunkIn.traceId; out.direction = chunkIn.direction; - out.frameKind = softbus::message_bus::pipeline::FrameKind::CompleteFrame; out.payloadSize = n; out.payload.block = std::move(blk); out.payload.offset = 0; diff --git a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h index a7b3dee..8a545b6 100644 --- a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h +++ b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h @@ -15,8 +15,8 @@ class PassthroughFramer final : public IFramer public: explicit PassthroughFramer(std::shared_ptr pool); - void feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, - std::vector& completeFramesOut) override; + void feed(softbus::core::models::RawBusMessage& chunkIn, + std::vector& completeFramesOut) override; private: std::shared_ptr m_pool; diff --git a/src/message_bus/pipeline/stages/2_parsers/IParserStage.h b/src/message_bus/pipeline/stages/2_parsers/IParserStage.h index b2ed7f2..cb49247 100644 --- a/src/message_bus/pipeline/stages/2_parsers/IParserStage.h +++ b/src/message_bus/pipeline/stages/2_parsers/IParserStage.h @@ -3,7 +3,7 @@ #include #include "core/models/DOMMessage.h" -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/RawBusMessage.h" namespace softbus::message_bus::pipeline::stages::parsers { @@ -13,7 +13,7 @@ class IParserStage public: virtual ~IParserStage() = default; virtual std::vector parse( - const softbus::message_bus::pipeline::PipelineContext& rawContext) = 0; + const softbus::core::models::RawBusMessage& rawContext) = 0; }; } // namespace softbus::message_bus::pipeline::stages::parsers diff --git a/src/message_bus/pipeline/stages/4_validators/IValidator.h b/src/message_bus/pipeline/stages/4_validators/IValidator.h index 19fea06..234f78b 100644 --- a/src/message_bus/pipeline/stages/4_validators/IValidator.h +++ b/src/message_bus/pipeline/stages/4_validators/IValidator.h @@ -1,6 +1,6 @@ #pragma once -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/RawBusMessage.h" namespace softbus::message_bus::pipeline::stages::validators { @@ -9,7 +9,7 @@ class IValidator { public: virtual ~IValidator() = default; - virtual bool validate(softbus::message_bus::pipeline::PipelineContext& ctx) = 0; + virtual bool validate(softbus::core::models::RawBusMessage& ctx) = 0; }; } // namespace softbus::message_bus::pipeline::stages::validators