modbus rtu 1.1 plugins

src/message_bus/pipeline/DynamicRuleRegistry.cpp

@@ -0,0 +1,95 @@
+#include "message_bus/pipeline/DynamicRuleRegistry.h"
+
+#include <mutex>
+#include <shared_mutex>
+
+namespace softbus::message_bus::pipeline
+{
+
+DynamicRuleRegistry& DynamicRuleRegistry::instance()
+{
+    static DynamicRuleRegistry registry;
+    return registry;
+}
+
+bool DynamicRuleRegistry::addFilterRule(const FilterRule& rule)
+{
+    if (rule.id.isEmpty()) {
+        return false;
+    }
+    std::unique_lock<std::shared_mutex> lk(m_mtx);
+    m_filterRules.insert(rule.id, rule);
+    return true;
+}
+
+bool DynamicRuleRegistry::removeFilterRule(const QString& id)
+{
+    std::unique_lock<std::shared_mutex> lk(m_mtx);
+    return m_filterRules.remove(id) > 0;
+}
+
+bool DynamicRuleRegistry::addTransformRule(const TransformRule& rule)
+{
+    if (rule.id.isEmpty()) {
+        return false;
+    }
+    std::unique_lock<std::shared_mutex> lk(m_mtx);
+    m_transformRules.insert(rule.id, rule);
+    return true;
+}
+
+bool DynamicRuleRegistry::removeTransformRule(const QString& id)
+{
+    std::unique_lock<std::shared_mutex> lk(m_mtx);
+    return m_transformRules.remove(id) > 0;
+}
+
+std::vector<FilterRule> DynamicRuleRegistry::snapshotFilterRules(
+    const QString& endpoint, int deviceId, const QString& stableKey) const
+{
+    std::shared_lock<std::shared_mutex> lk(m_mtx);
+    std::vector<FilterRule> out;
+    out.reserve(static_cast<std::size_t>(m_filterRules.size()));
+    for (auto it = m_filterRules.constBegin(); it != m_filterRules.constEnd(); ++it) {
+        if (matchScope(it.value().scope, endpoint, deviceId, stableKey)) {
+            out.push_back(it.value());
+        }
+    }
+    return out;
+}
+
+std::vector<TransformRule> DynamicRuleRegistry::snapshotTransformRules(
+    const QString& endpoint, int deviceId, const QString& stableKey) const
+{
+    std::shared_lock<std::shared_mutex> lk(m_mtx);
+    std::vector<TransformRule> out;
+    out.reserve(static_cast<std::size_t>(m_transformRules.size()));
+    for (auto it = m_transformRules.constBegin(); it != m_transformRules.constEnd(); ++it) {
+        if (matchScope(it.value().scope, endpoint, deviceId, stableKey)) {
+            out.push_back(it.value());
+        }
+    }
+    return out;
+}
+
+QJsonObject DynamicRuleRegistry::status() const
+{
+    std::shared_lock<std::shared_mutex> lk(m_mtx);
+    QJsonObject o;
+    o.insert(QStringLiteral("filterRuleCount"), m_filterRules.size());
+    o.insert(QStringLiteral("transformRuleCount"), m_transformRules.size());
+    return o;
+}
+
+bool DynamicRuleRegistry::matchScope(
+    const RuleScope& scope, const QString& endpoint, int deviceId, const QString& stableKey) const
+{
+    // hybrid scope: global + endpoint/device/stableKey selective match
+    const bool endpointOk = scope.endpoint.isEmpty() || scope.endpoint == endpoint;
+    const bool deviceOk = scope.deviceId < 0 || scope.deviceId == deviceId;
+    const bool stableKeyOk = scope.stableKey.isEmpty() || scope.stableKey == stableKey;
+    return endpointOk && deviceOk && stableKeyOk;
+}
+
+} // namespace softbus::message_bus::pipeline
+

src/message_bus/pipeline/DynamicRuleRegistry.h

@@ -0,0 +1,64 @@
+#pragma once
+
+#include <shared_mutex>
+#include <vector>
+
+#include <QHash>
+#include <QJsonObject>
+#include <QString>
+
+namespace softbus::message_bus::pipeline
+{
+
+struct RuleScope
+{
+    QString endpoint;
+    int deviceId{-1};
+    QString stableKey;
+};
+
+struct FilterRule
+{
+    QString id;
+    RuleScope scope;
+    QJsonObject condition;
+};
+
+struct TransformRule
+{
+    QString id;
+    RuleScope scope;
+    double scale{1.0};
+    double offset{0.0};
+    QString unit;
+};
+
+class DynamicRuleRegistry
+{
+public:
+    static DynamicRuleRegistry& instance();
+
+    bool addFilterRule(const FilterRule& rule);
+    bool removeFilterRule(const QString& id);
+    bool addTransformRule(const TransformRule& rule);
+    bool removeTransformRule(const QString& id);
+
+    std::vector<FilterRule> snapshotFilterRules(
+        const QString& endpoint, int deviceId, const QString& stableKey) const;
+    std::vector<TransformRule> snapshotTransformRules(
+        const QString& endpoint, int deviceId, const QString& stableKey) const;
+
+    QJsonObject status() const;
+
+private:
+    DynamicRuleRegistry() = default;
+    bool matchScope(const RuleScope& scope, const QString& endpoint, int deviceId, const QString& stableKey) const;
+
+private:
+    mutable std::shared_mutex m_mtx;
+    QHash<QString, FilterRule> m_filterRules;
+    QHash<QString, TransformRule> m_transformRules;
+};
+
+} // namespace softbus::message_bus::pipeline
+

src/message_bus/pipeline/PipelineContext.h

@@ -5,14 +5,10 @@
 #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
 {
@@ -46,13 +42,6 @@ struct PipelineContext
     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

@@ -5,9 +5,6 @@
 #include <sstream>
 #include <chrono>
 #include <functional>
-
-#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h"
-#include "plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h"
 #include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
 #include "utils/logs/logging.h"
 
@@ -91,9 +88,6 @@ void PipelineEngine::loadRuntimeConfigFromJson(const QJsonObject& obj)
         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.modbusMaxAdu              = static_cast<std::size_t>(
-        std::max(8, obj.value(QStringLiteral("modbusRtuMaxFrameBytes")).toInt(256)));
-    m_rtConfig.modbusInterFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0);
 }
 
 void PipelineEngine::start()
@@ -110,13 +104,6 @@ void PipelineEngine::start()
 
     m_passthroughFramer =
         std::make_shared<softbus::message_bus::pipeline::stages::framers::PassthroughFramer>(m_pool);
-    if (m_pluginManager) {
-        softbus::message_bus::pipeline::protocol::modbusrtu::ModbusRtuFramerPluginConfig cfg;
-        cfg.maxAdu = m_rtConfig.modbusMaxAdu;
-        cfg.interFrameTimeoutMs = m_rtConfig.modbusInterFrameTimeoutMs;
-        m_pluginManager->registerProtocolFramerPlugin(
-            softbus::message_bus::pipeline::protocol::modbusrtu::makeModbusRtuFramerPlugin(m_pool, cfg));
-    }
 
     // 如果启用并行管道 则创建帧队列和解析线程 如果启用有序出口 则创建合并队列和合并线程
     // 并行管道是用来处理多个端点的数据流的
@@ -183,6 +170,14 @@ void PipelineEngine::stop()
         std::lock_guard<std::mutex> lk(m_framerSessionMutex);
         m_framerSessions.clear();
     }
+    {
+        std::lock_guard<std::mutex> lk(m_parserSessionMutex);
+        m_parserSessions.clear();
+    }
+    {
+        std::lock_guard<std::mutex> lk(m_mapperSessionMutex);
+        m_mapperSessions.clear();
+    }
 }
 // 功能 ：清空 ingress 队列
 // 参数 ：无
@@ -294,10 +289,7 @@ void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector<Pipeli
             std::lock_guard<std::mutex> lk(m_framerSessionMutex);
             auto it = m_framerSessions.find(key);
             if (it == m_framerSessions.end() || !it.value()) {
-                QJsonObject params;
-                params.insert(QStringLiteral("modbusRtuMaxFrameBytes"), static_cast<int>(m_rtConfig.modbusMaxAdu));
-                params.insert(QStringLiteral("modbusRtuInterFrameTimeoutMs"), m_rtConfig.modbusInterFrameTimeoutMs);
-                auto s = plugin->createFramerSession(chunkIn.endpoint, params);
+                auto s = plugin->createFramerSession(chunkIn.endpoint, QJsonObject{});
                 m_framerSessions.insert(
                     key, s ? std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>(std::move(s)) : nullptr);
                 it = m_framerSessions.find(key);
@@ -454,44 +446,19 @@ bool PipelineEngine::runStages234(PipelineContext& ctx)
     if (ctx.frameKind != FrameKind::CompleteFrame) {
         return false;
     }
-    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 << ' ';
+    auto messages = stage2Parser(ctx);
+    LOG_INFO() << "PipelineEngine: stage2Parser messages size=" << messages.size();
+    if (messages.empty()) {
+        return false;
+    }
+    for (auto& msg : messages) {
+        if (!stage3Filter(msg)) {
+            continue;
         }
-    }
-    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");
-    }
-    // @TODO 以后需要测试是否丢包和解析延迟
-    // m_count++;
-    // LOG_INFO() << "the count of command is" << m_count;
-
-    if (!stage3Filter(ctx)) {
-        return false;
-    }
-    if (!stage4Validator(ctx)) {
-        return false;
+        if (!stage4Validator(msg)) {
+            continue;
+        }
+        (void)applyTransformAndEnrich(ctx, msg);
     }
     return true;
 }
@@ -500,47 +467,97 @@ bool PipelineEngine::runStages234(PipelineContext& ctx)
 // 参数 ：ctx 数据
 // 返回值 ：是否成功
 // 逻辑 ：如果有效载荷无效 则返回false 如果都成功 则返回true
-bool PipelineEngine::stage4Validator(PipelineContext& ctx)
+bool PipelineEngine::stage4Validator(const softbus::core::models::DOMMessage& message) const
 {
-    if (!ctx.payload.valid()) {
-        return false;
-    }
-    if (ctx.endpoint.isEmpty()) {
-        return false;
-    }
-    if (!ctx.timestamp.isValid()) {
-        ctx.timestamp = QDateTime::currentDateTimeUtc();
-    }
-    if (ctx.frameKind != FrameKind::CompleteFrame) {
-        return false;
-    }
-    return true;
+    return !message.messageId.isEmpty();
 }
 
 // 功能 ：运行阶段2解析器
 // 参数 ：ctx 数据
 // 返回值 ：是否成功
 // 逻辑 ：如果插件管理器为空 则返回true 如果插件管理器不为空 则选择合适的插件 创建会话 如果会话为空 则返回false 如果会话不为空 则返回true
-bool PipelineEngine::stage2Parser(PipelineContext& ctx)
+std::vector<softbus::core::models::DOMMessage> PipelineEngine::stage2Parser(const PipelineContext& ctx)
 {
     if (!m_pluginManager) {
-        return true;
+        return {};
     }
-    auto plugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint);
-    if (!plugin) {
-        return true;
+    auto parserPlugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint);
+    if (!parserPlugin) {
+        return {};
     }
-    auto session = plugin->createSession();
-    if (!session) {
-        return false;
+    const QString parserKey = ctx.endpoint + QStringLiteral("|") + parserPlugin->pluginId();
+    std::shared_ptr<softbus::core::plugin_system::IProtocolSession> parserSession;
+    {
+        std::lock_guard<std::mutex> lk(m_parserSessionMutex);
+        auto it = m_parserSessions.find(parserKey);
+        if (it == m_parserSessions.end() || !it.value()) {
+            auto s = parserPlugin->createSession();
+            m_parserSessions.insert(
+                parserKey,
+                s ? std::shared_ptr<softbus::core::plugin_system::IProtocolSession>(std::move(s)) : nullptr);
+            it = m_parserSessions.find(parserKey);
+        }
+        if (it == m_parserSessions.end() || !it.value()) {
+            return {};
+        }
+        parserSession = it.value();
     }
-    return session->feed(ctx);
+    softbus::message_bus::pipeline::protocol::ProtocolEnvelope envelope; // 协议封装
+    if (!parserSession->parse(ctx, envelope)) {
+        return {};
+    }
+
+    auto mapperPlugin = m_pluginManager->selectProtocolMapperPlugin(ctx.protocolHint);
+    if (!mapperPlugin) {
+        return {};
+    }
+    const QString key = ctx.endpoint + QStringLiteral("|") + mapperPlugin->pluginId();
+    std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession> session;
+    {
+        std::lock_guard<std::mutex> lk(m_mapperSessionMutex);
+        auto it = m_mapperSessions.find(key);
+        if (it == m_mapperSessions.end() || !it.value()) {
+            auto s = mapperPlugin->createMapperSession();
+            m_mapperSessions.insert(
+                key,
+                s ? std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>(std::move(s)) : nullptr);
+            it = m_mapperSessions.find(key);
+        }
+        if (it == m_mapperSessions.end() || !it.value()) {
+            return {};
+        }
+        session = it.value();
+    }
+    return session->map(ctx, envelope);
 }
 
-bool PipelineEngine::stage3Filter(PipelineContext& ctx)
+bool PipelineEngine::stage3Filter(softbus::core::models::DOMMessage& message)
 {
     static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
-    return s_filter.apply(ctx);
+    return s_filter.process(message);
+}
+
+softbus::core::models::EnrichedMessage PipelineEngine::applyTransformAndEnrich(
+    const PipelineContext& rawCtx, softbus::core::models::DOMMessage& message) const
+{
+    auto rules = softbus::message_bus::pipeline::DynamicRuleRegistry::instance().snapshotTransformRules(
+        rawCtx.endpoint, rawCtx.deviceId, rawCtx.stableKey);
+    for (const auto& rule : rules) {
+        const double base = std::visit([](const auto& x) { return static_cast<double>(x); }, message.value);
+        const double next = base * rule.scale + rule.offset;
+        if (next != base) {
+            message.appendTrace(QStringLiteral("transform:%1 %2->%3")
+                                    .arg(rule.id)
+                                    .arg(base, 0, 'g', 12)
+                                    .arg(next, 0, 'g', 12));
+        }
+        message.value = next;
+        if (!rule.unit.isEmpty()) {
+            message.attributes.insert(QStringLiteral("unit"), rule.unit);
+        }
+    }
+    static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
+    return s_filter.enrich(message);
 }
 
 } // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/PipelineEngine.h

@@ -12,9 +12,13 @@
 
 #include "core/plugin_system/PluginManager.h"
 #include "core/plugin_system/IProtocolFramerPlugin.h"
+#include "core/plugin_system/IProtocolMapperPlugin.h"
+#include "core/models/DOMMessage.h"
+#include "core/models/EnrichedMessage.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"
 
@@ -29,9 +33,6 @@ struct PipelineRuntimeConfig
 
     std::size_t framedQueueMax{4096};
     std::size_t maxReorderDepth{256};
-
-    std::size_t modbusMaxAdu{256};
-    int modbusInterFrameTimeoutMs{0};
 };
 
 struct EndpointFramingState;
@@ -92,9 +93,11 @@ private:
     bool runStages234(PipelineContext& ctx);
     void ingestMergeItem(MergeItem item);
 
-    bool stage2Parser(PipelineContext& ctx);
-    bool stage3Filter(PipelineContext& ctx);
-    bool stage4Validator(PipelineContext& ctx);
+    std::vector<softbus::core::models::DOMMessage> stage2Parser(const PipelineContext& 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;
 
 private:
     softbus::core::memory::LockFreeQueue<PipelineContext> m_ingressQ;
@@ -110,6 +113,10 @@ private:
     std::shared_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> m_passthroughFramer;
     std::mutex m_framerSessionMutex;
     QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>> m_framerSessions;
+    std::mutex m_parserSessionMutex;
+    QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolSession>> m_parserSessions;
+    std::mutex m_mapperSessionMutex;
+    QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>> m_mapperSessions;
 
     std::mutex                    m_seqMutex;
     QHash<QString, std::uint64_t> m_nextFrameSeqByEndpoint;

src/message_bus/pipeline/protocol/ProtocolEnvelope.h

@@ -1,8 +1,9 @@
 #pragma once
 
+#include <cstdint>
 #include <variant>
 
-#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h"
+#include "plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h"
 
 namespace softbus::message_bus::pipeline::protocol
 {
@@ -11,10 +12,16 @@ enum class ProtocolFamily : std::uint8_t
 {
     None = 0,
     ModbusRtu,
+    CanOpen,
 };
 
-/// 流水线内协议解析结果的统一外壳；后续可在此 variant 中增加 CAN 等协议。
-using ProtocolEnvelopePdu = std::variant<std::monostate, ModbusRtuPdu>;
+/// CANopen 协议占位类型（后续接入真实 CANopen 对象后替换/扩展）。
+struct CanOpenPduPlaceholder
+{
+};
+
+/// 流水线内协议解析结果的统一外壳；通过 variant 承载多协议强类型结果。
+using ProtocolEnvelopePdu = std::variant<std::monostate, ModbusRtuPdu, CanOpenPduPlaceholder>;
 
 struct ProtocolEnvelope
 {

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

@@ -22,7 +22,6 @@ void PassthroughFramer::feed(softbus::message_bus::pipeline::PipelineContext&
     }
     auto blk = m_pool->allocate(n);
     if (!blk || !blk->data || blk->capacity < n) {
-        chunkIn.framerError = QStringLiteral("pool_exhausted");
         return;
     }
     const std::uint8_t* src = chunkIn.payload.bytes();

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

@@ -1,5 +1,8 @@
 #pragma once
 // 定义解析器阶段接口
+#include <vector>
+
+#include "core/models/DOMMessage.h"
 #include "message_bus/pipeline/PipelineContext.h"
 
 namespace softbus::message_bus::pipeline::stages::parsers
@@ -9,7 +12,8 @@ class IParserStage
 {
 public:
     virtual ~IParserStage() = default;
-    virtual bool parse(softbus::message_bus::pipeline::PipelineContext& ctx) = 0;
+    virtual std::vector<softbus::core::models::DOMMessage> parse(
+        const softbus::message_bus::pipeline::PipelineContext& rawContext) = 0;
 };
 
 } // namespace softbus::message_bus::pipeline::stages::parsers

src/message_bus/pipeline/stages/3_filters/IFilterStage.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "message_bus/pipeline/PipelineContext.h"
+#include "core/models/DOMMessage.h"
 
 namespace softbus::message_bus::pipeline::stages::filters
 {
@@ -9,7 +9,7 @@ class IFilterStage
 {
 public:
     virtual ~IFilterStage() = default;
-    virtual bool apply(softbus::message_bus::pipeline::PipelineContext& ctx) = 0;
+    virtual bool process(softbus::core::models::DOMMessage& message) = 0;
 };
 
 } // namespace softbus::message_bus::pipeline::stages::filters

src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp

@@ -1,565 +1,67 @@
 #include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
 
-#include <algorithm>
-#include <array>
-#include <cctype>
-#include <cmath>
-#include <cstring>
-#include <limits>
-#include <optional>
-
 #include <QDir>
-#include <QFile>
 #include <QFileInfo>
-#include <QJsonArray>
-#include <QVector>
-
-#include <variant>
-
-#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h"
-#include "message_bus/pipeline/protocol/ProtocolEnvelope.h"
-#include "utils/logs/logging.h"
 
 namespace softbus::message_bus::pipeline::stages::filters
 {
 
-namespace
-{
-
-using softbus::core::models::DOMMessage;
-using softbus::core::models::DataPointMapping;
-using softbus::core::models::DataQuality;
-using softbus::core::models::DataType;
-using softbus::core::models::Endianness;
-using softbus::message_bus::pipeline::protocol::ModbusRtuPdu;
-using softbus::message_bus::pipeline::protocol::ProtocolFamily;
-using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse;
-
-QByteArray registersToBytes(const QVector<std::uint16_t>& regs)
-{
-    QByteArray out;
-    out.reserve(regs.size() * 2);
-    for (std::uint16_t v : regs) {
-        out.append(static_cast<char>((v >> 8) & 0xFF));
-        out.append(static_cast<char>(v & 0xFF));
-    }
-    return out;
-}
-
-QJsonArray registersToQJsonArray(const QVector<std::uint16_t>& regs)
-{
-    QJsonArray a;
-    for (std::uint16_t v : regs) {
-        a.append(static_cast<int>(v));
-    }
-    return a;
-}
-
-std::uint64_t readUnsignedRaw(const QByteArray& bytes, const DataPointMapping& mapping, bool* ok)
-{
-    *ok = false;
-    if (mapping.byteLength == 0 || mapping.byteLength > 8) {
-        return 0;
-    }
-    const int offset = static_cast<int>(mapping.byteOffset);
-    const int len    = static_cast<int>(mapping.byteLength);
-    if (offset < 0 || offset + len > bytes.size()) {
-        return 0;
-    }
-
-    std::array<std::uint8_t, 8> buf{};
-    for (int i = 0; i < len; ++i) {
-        buf[static_cast<std::size_t>(i)] = static_cast<std::uint8_t>(bytes[offset + i]);
-    }
-
-    if (mapping.endianness == Endianness::LittleEndian) {
-        std::reverse(buf.begin(), buf.begin() + len);
-    } else if (mapping.endianness == Endianness::BigEndianByteSwap && (len % 2 == 0)) {
-        for (int i = 0; i + 1 < len; i += 2) {
-            std::swap(buf[static_cast<std::size_t>(i)], buf[static_cast<std::size_t>(i + 1)]);
-        }
-    }
-
-    std::uint64_t v = 0;
-    for (int i = 0; i < len; ++i) {
-        v = (v << 8) | buf[static_cast<std::size_t>(i)];
-    }
-    *ok = true;
-    return v;
-}
-
-double decodeRawToDouble(std::uint64_t raw, const DataPointMapping& mapping)
-{
-    switch (mapping.rawType) {
-    case DataType::BOOL:
-        return raw ? 1.0 : 0.0;
-    case DataType::INT32:
-        return static_cast<double>(static_cast<std::int32_t>(raw & 0xFFFFFFFFu));
-    case DataType::UINT32:
-        return static_cast<double>(static_cast<std::uint32_t>(raw & 0xFFFFFFFFu));
-    case DataType::INT64:
-        return static_cast<double>(static_cast<std::int64_t>(raw));
-    case DataType::FLOAT32: {
-        const std::uint32_t u = static_cast<std::uint32_t>(raw & 0xFFFFFFFFu);
-        float               f = 0.0f;
-        std::memcpy(&f, &u, sizeof(float));
-        return static_cast<double>(f);
-    }
-    case DataType::FLOAT64: {
-        double d = 0.0;
-        std::memcpy(&d, &raw, sizeof(double));
-        return d;
-    }
-    case DataType::STRING:
-    case DataType::BYTE_ARRAY:
-    default:
-        return static_cast<double>(raw);
-    }
-}
-
-class ExprParser
-{
-public:
-    ExprParser(const QString& expr, const QJsonArray& regs, double x)
-        : m_expr(expr)
-        , m_regs(regs)
-        , m_x(x)
-    {
-    }
-
-    bool eval(double* out)
-    {
-        m_pos = 0;
-        if (!out) {
-            return false;
-        }
-        const auto v = parseBitOr();
-        skipSpaces();
-        if (!v.has_value() || m_pos != m_expr.size()) {
-            return false;
-        }
-        *out = *v;
-        return true;
-    }
-
-private:
-    std::optional<double> parseBitOr()
-    {
-        auto lhs = parseBitAnd();
-        if (!lhs.has_value()) {
-            return std::nullopt;
-        }
-        while (true) {
-            skipSpaces();
-            if (!match('|')) {
-                break;
-            }
-            auto rhs = parseBitAnd();
-            if (!rhs.has_value()) {
-                return std::nullopt;
-            }
-            const auto lv = static_cast<std::int64_t>(*lhs);
-            const auto rv = static_cast<std::int64_t>(*rhs);
-            lhs           = static_cast<double>(lv | rv);
-        }
-        return lhs;
-    }
-
-    std::optional<double> parseBitAnd()
-    {
-        auto lhs = parseShift();
-        if (!lhs.has_value()) {
-            return std::nullopt;
-        }
-        while (true) {
-            skipSpaces();
-            if (!match('&')) {
-                break;
-            }
-            auto rhs = parseShift();
-            if (!rhs.has_value()) {
-                return std::nullopt;
-            }
-            const auto lv = static_cast<std::int64_t>(*lhs);
-            const auto rv = static_cast<std::int64_t>(*rhs);
-            lhs           = static_cast<double>(lv & rv);
-        }
-        return lhs;
-    }
-
-    std::optional<double> parseShift()
-    {
-        auto lhs = parseAddSub();
-        if (!lhs.has_value()) {
-            return std::nullopt;
-        }
-        while (true) {
-            skipSpaces();
-            if (match(QStringLiteral("<<"))) {
-                auto rhs = parseAddSub();
-                if (!rhs.has_value()) {
-                    return std::nullopt;
-                }
-                lhs = static_cast<double>(static_cast<std::int64_t>(*lhs) << static_cast<int>(*rhs));
-                continue;
-            }
-            if (match(QStringLiteral(">>"))) {
-                auto rhs = parseAddSub();
-                if (!rhs.has_value()) {
-                    return std::nullopt;
-                }
-                lhs = static_cast<double>(static_cast<std::int64_t>(*lhs) >> static_cast<int>(*rhs));
-                continue;
-            }
-            break;
-        }
-        return lhs;
-    }
-
-    std::optional<double> parseAddSub()
-    {
-        auto lhs = parseMulDiv();
-        if (!lhs.has_value()) {
-            return std::nullopt;
-        }
-        while (true) {
-            skipSpaces();
-            if (match('+')) {
-                auto rhs = parseMulDiv();
-                if (!rhs.has_value()) {
-                    return std::nullopt;
-                }
-                lhs = *lhs + *rhs;
-                continue;
-            }
-            if (match('-')) {
-                auto rhs = parseMulDiv();
-                if (!rhs.has_value()) {
-                    return std::nullopt;
-                }
-                lhs = *lhs - *rhs;
-                continue;
-            }
-            break;
-        }
-        return lhs;
-    }
-
-    std::optional<double> parseMulDiv()
-    {
-        auto lhs = parseUnary();
-        if (!lhs.has_value()) {
-            return std::nullopt;
-        }
-        while (true) {
-            skipSpaces();
-            if (match('*')) {
-                auto rhs = parseUnary();
-                if (!rhs.has_value()) {
-                    return std::nullopt;
-                }
-                lhs = *lhs * *rhs;
-                continue;
-            }
-            if (match('/')) {
-                auto rhs = parseUnary();
-                if (!rhs.has_value() || *rhs == 0.0) {
-                    return std::nullopt;
-                }
-                lhs = *lhs / *rhs;
-                continue;
-            }
-            break;
-        }
-        return lhs;
-    }
-
-    std::optional<double> parseUnary()
-    {
-        skipSpaces();
-        if (match('-')) {
-            auto v = parseUnary();
-            return v.has_value() ? std::optional<double>(-*v) : std::nullopt;
-        }
-        if (match('+')) {
-            return parseUnary();
-        }
-        return parsePrimary();
-    }
-
-    std::optional<double> parsePrimary()
-    {
-        skipSpaces();
-        if (match('(')) {
-            auto v = parseBitOr();
-            skipSpaces();
-            if (!v.has_value() || !match(')')) {
-                return std::nullopt;
-            }
-            return v;
-        }
-
-        if (peekIsAlpha()) {
-            return parseVariable();
-        }
-        return parseNumber();
-    }
-
-    std::optional<double> parseVariable()
-    {
-        const auto name = parseIdentifier();
-        if (name == QStringLiteral("x")) {
-            return m_x;
-        }
-        if (name.startsWith(QStringLiteral("r"))) {
-            bool ok = false;
-            const int idx = name.mid(1).toInt(&ok);
-            if (ok && idx >= 0 && idx < m_regs.size()) {
-                return static_cast<double>(m_regs[idx].toInt(0));
-            }
-        }
-        return std::nullopt;
-    }
-
-    std::optional<double> parseNumber()
-    {
-        skipSpaces();
-        const int start = m_pos;
-        bool      dot   = false;
-        while (m_pos < m_expr.size()) {
-            const QChar c = m_expr[m_pos];
-            if (c.isDigit()) {
-                ++m_pos;
-                continue;
-            }
-            if (c == QChar('.')) {
-                if (dot) {
-                    break;
-                }
-                dot = true;
-                ++m_pos;
-                continue;
-            }
-            break;
-        }
-        if (m_pos <= start) {
-            return std::nullopt;
-        }
-        bool ok = false;
-        const double v = m_expr.mid(start, m_pos - start).toDouble(&ok);
-        if (!ok) {
-            return std::nullopt;
-        }
-        return v;
-    }
-
-    QString parseIdentifier()
-    {
-        skipSpaces();
-        const int start = m_pos;
-        while (m_pos < m_expr.size()) {
-            const QChar c = m_expr[m_pos];
-            if (c.isLetterOrNumber() || c == QChar('_')) {
-                ++m_pos;
-                continue;
-            }
-            break;
-        }
-        return m_expr.mid(start, m_pos - start);
-    }
-
-    bool match(QChar c)
-    {
-        if (m_pos < m_expr.size() && m_expr[m_pos] == c) {
-            ++m_pos;
-            return true;
-        }
-        return false;
-    }
-
-    bool match(const QString& token)
-    {
-        if (m_expr.mid(m_pos, token.size()) == token) {
-            m_pos += token.size();
-            return true;
-        }
-        return false;
-    }
-
-    bool peekIsAlpha() const
-    {
-        return m_pos < m_expr.size() && (m_expr[m_pos].isLetter() || m_expr[m_pos] == QChar('_'));
-    }
-
-    void skipSpaces()
-    {
-        while (m_pos < m_expr.size() && m_expr[m_pos].isSpace()) {
-            ++m_pos;
-        }
-    }
-
-private:
-    QString    m_expr;
-    QJsonArray m_regs;
-    double     m_x{0.0};
-    int        m_pos{0};
-};
-
-softbus::core::models::DOMValue castToSemantic(double value, DataType t)
-{
-    using softbus::core::models::DOMValue;
-    switch (t) {
-    case DataType::BOOL:
-        return DOMValue{value != 0.0};
-    case DataType::INT32:
-        return DOMValue{static_cast<std::int32_t>(value)};
-    case DataType::UINT32:
-        return DOMValue{static_cast<std::uint32_t>(value)};
-    case DataType::INT64:
-        return DOMValue{static_cast<std::int64_t>(value)};
-    case DataType::FLOAT32:
-        return DOMValue{static_cast<float>(value)};
-    case DataType::FLOAT64:
-    default:
-        return DOMValue{value};
-    }
-}
-
-DataQuality qualityByRange(double value, const softbus::core::models::MetadataDef& meta)
-{
-    if (!std::isfinite(value)) {
-        return DataQuality::BAD;
-    }
-    if (meta.maxValue > meta.minValue && (value < meta.minValue || value > meta.maxValue)) {
-        return DataQuality::UNCERTAIN;
-    }
-    return DataQuality::GOOD;
-}
-
-QString normalizedProtocol(const QString& protocolHint)
-{
-    return protocolHint.trimmed().toLower();
-}
-
-} // namespace
-
 ProtocolParseFilter::ProtocolParseFilter() = default;
 
-bool ProtocolParseFilter::apply(softbus::message_bus::pipeline::PipelineContext& ctx)
+bool ProtocolParseFilter::process(softbus::core::models::DOMMessage& message)
 {
-    if (ctx.direction == softbus::message_bus::pipeline::BusDirection::Downstream) {
+    if (!ensureLoaded()) {
         return true;
     }
-    if (!ensureLoaded()) {
-        return true; // 保持兼容：配置缺失不阻断流水线
+    // Message-stage filter: only normalize quality by metadata range.
+    if (!message.metaRef) {
+        return true;
     }
-    if (normalizedProtocol(ctx.protocolHint).contains(QStringLiteral("modbus"))) {
-        return parseModbus(ctx);
+    const auto* meta = m_metadataRegistry.findById(message.metaRef->metadataId);
+    if (!meta) {
+        return true;
+    }
+    const double v = std::visit(
+        [](const auto& x) { return static_cast<double>(x); },
+        message.value);
+    if (meta->maxValue > meta->minValue && (v < meta->minValue || v > meta->maxValue)) {
+        message.quality = softbus::core::models::DataQuality::UNCERTAIN;
+        message.appendTrace(QStringLiteral("filter:range_check -> uncertain"));
     }
     return true;
 }
 
-bool ProtocolParseFilter::ensureLoaded()
+softbus::core::models::EnrichedMessage
+ProtocolParseFilter::enrich(const softbus::core::models::DOMMessage& message)
+{
+    softbus::core::models::EnrichedMessage out;
+    out.sourceMessageId = message.messageId;
+    out.logicalName = message.metaRef ? message.metaRef->name : message.messageId;
+    out.unit = message.metaRef ? message.metaRef->unit : QString();
+    out.engineeringValue = message.value;
+    out.quality = message.quality;
+    out.attributes = message.attributes;
+    out.traceLog = message.traceLog;
+    return out;
+}
+
+bool ProtocolParseFilter::ensureLoaded() const
 {
     std::call_once(m_loadOnce, [this]() {
         QString configBase = qEnvironmentVariable("SOFTBUS_CONFIG_DIR");
         if (!configBase.isEmpty() && !configBase.endsWith(QChar('/'))) {
             configBase.append(QChar('/'));
         }
-        const QString homeCfg   = configBase.isEmpty()
-                                     ? (QDir::homePath() + QStringLiteral("/softbus/config/"))
-                                     : configBase;
-        const QString metaHome  = homeCfg + QStringLiteral("metadata_dictionary.json");
-        const QString profHome  = homeCfg + QStringLiteral("device_profiles.json");
+        const QString homeCfg = configBase.isEmpty()
+            ? (QDir::homePath() + QStringLiteral("/softbus/config/"))
+            : configBase;
+        const QString metaHome = homeCfg + QStringLiteral("metadata_dictionary.json");
         const QString metaLocal = QStringLiteral("config/metadata_dictionary.json");
-        const QString profLocal = QStringLiteral("config/device_profiles.json");
-
         const QString metaPath = QFileInfo::exists(metaHome) ? metaHome : metaLocal;
-        const QString profPath = QFileInfo::exists(profHome) ? profHome : profLocal;
-
-        const bool metaOk = m_metadataRegistry.loadFromFile(metaPath);
-        const bool profOk = m_profileRegistry.loadFromFile(profPath);
-        if (!metaOk || !profOk) {
-            LOG_WARNING() << "ProtocolParseFilter: config load failed metadata="
-                          << m_metadataRegistry.lastError() << " profile="
-                          << m_profileRegistry.lastError();
-            m_loaded = false;
-            return;
-        }
-        m_loaded = true;
+        m_loaded = m_metadataRegistry.loadFromFile(metaPath);
     });
     return m_loaded;
 }
 
-bool ProtocolParseFilter::parseModbus(softbus::message_bus::pipeline::PipelineContext& ctx)
-{
-    QVector<std::uint16_t> regsVec;
-    QJsonArray             regsForExpr;
-
-    if (ctx.protocol.family == ProtocolFamily::ModbusRtu) {
-        if (const auto* m = std::get_if<ModbusRtuPdu>(&ctx.protocol.pdu)) {
-            if (const auto* resp = std::get_if<ReadHoldingRegistersResponse>(&*m)) {
-                regsVec     = resp->registers;
-                regsForExpr = registersToQJsonArray(regsVec);
-            }
-        }
-    }
-    if (regsVec.isEmpty()) {
-        const QJsonArray regsLegacy = ctx.parsed.value(QStringLiteral("registers")).toArray();
-        if (regsLegacy.isEmpty()) {
-            return true;
-        }
-        regsForExpr = regsLegacy;
-        for (const auto& reg : regsLegacy) {
-            regsVec.append(static_cast<std::uint16_t>(reg.toInt(0) & 0xFFFF));
-        }
-    }
-
-    const auto* profile = m_profileRegistry.find(ctx.stableKey, ctx.protocolHint);
-    if (!profile) {
-        profile = m_profileRegistry.find(QString::number(ctx.deviceId), ctx.protocolHint);
-    }
-    if (!profile) {
-        return true;
-    }
-
-    const QByteArray bytes = registersToBytes(regsVec);
-    ctx.domMessages.clear();
-    ctx.domMessages.reserve(profile->mappings.size());
-
-    for (const auto& mapping : profile->mappings) {
-        const auto* meta = m_metadataRegistry.findById(mapping.metadataId);
-        if (!meta) {
-            continue;
-        }
-
-        bool ok = false;
-        const std::uint64_t raw = readUnsignedRaw(bytes, mapping, &ok);
-        if (!ok) {
-            continue;
-        }
-
-        double v = decodeRawToDouble(raw, mapping);
-        v        = v * mapping.scaleFactor + mapping.offsetFactor;
-        if (!mapping.expression.trimmed().isEmpty()) {
-            ExprParser parser(mapping.expression, regsForExpr, v);
-            double     exprV = 0.0;
-            if (parser.eval(&exprV)) {
-                v = exprV;
-            }
-        }
-
-        DOMMessage msg;
-        msg.messageId = QStringLiteral("%1.%2")
-                            .arg(ctx.stableKey.isEmpty() ? QString::number(ctx.deviceId) : ctx.stableKey,
-                                 mapping.pointId);
-        msg.metaRef   = meta;
-        msg.timestamp = ctx.timestamp.toMSecsSinceEpoch();
-        msg.value     = castToSemantic(v, meta->semanticType);
-        msg.quality   = qualityByRange(v, *meta);
-        ctx.domMessages.append(std::move(msg));
-    }
-    return true;
-}
-
 } // namespace softbus::message_bus::pipeline::stages::filters
+

src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h

@@ -2,8 +2,8 @@
 
 #include <mutex>
 
+#include "core/models/EnrichedMessage.h"
 #include "core/metadata/MetadataRegistry.h"
-#include "core/metadata/ProfileRegistry.h"
 #include "message_bus/pipeline/stages/3_filters/IFilterStage.h"
 
 namespace softbus::message_bus::pipeline::stages::filters
@@ -14,17 +14,16 @@ class ProtocolParseFilter final : public IFilterStage
 public:
     ProtocolParseFilter();
 
-    bool apply(softbus::message_bus::pipeline::PipelineContext& ctx) override;
+    bool process(softbus::core::models::DOMMessage& message) override;
+    softbus::core::models::EnrichedMessage enrich(const softbus::core::models::DOMMessage& message);
 
 private:
-    bool ensureLoaded();
-    bool parseModbus(softbus::message_bus::pipeline::PipelineContext& ctx);
+    bool ensureLoaded() const;
 
 private:
-    softbus::core::metadata::MetadataRegistry m_metadataRegistry;
-    softbus::core::metadata::ProfileRegistry  m_profileRegistry;
-    std::once_flag                            m_loadOnce;
-    bool                                      m_loaded{false};
+    mutable softbus::core::metadata::MetadataRegistry m_metadataRegistry;
+    mutable std::once_flag                            m_loadOnce;
+    mutable bool                                      m_loaded{false};
 };
 
 } // namespace softbus::message_bus::pipeline::stages::filters