endpointhash 1.0

src/message_bus/pipeline/PipelineEngine.cpp

@@ -6,12 +6,15 @@
 #include <chrono>
 #include <functional>
 #include "core/models/MessageRoutingUtils.h"
+#include "message_bus/pipeline/discovery/DiscoveryPool.h"
+#include "message_bus/pipeline/mapping/MappingRegistry.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<softbus::core::models::RawBusMessage> chunkQ;
@@ -32,6 +35,41 @@ struct EndpointFramingState
 namespace
 {
 
+class MappingLookupAdapter final : public softbus::core::plugin_system::IMappingLookup
+{
+public:
+    std::optional<softbus::core::plugin_system::MappingLookupResult> find(std::uint64_t compositeKey) const override
+    {
+        auto rule = softbus::message_bus::pipeline::MappingRegistry::instance().find(compositeKey);
+        if (!rule.has_value()) {
+            return std::nullopt;
+        }
+        softbus::core::plugin_system::MappingLookupResult out;
+        out.domPath = rule->domPath;
+        out.scale = rule->scale;
+        out.offset = rule->offset;
+        out.unit = rule->unit;
+        return out;
+    }
+};
+
+class DiscoverySinkAdapter final : public softbus::core::plugin_system::IDiscoverySink
+{
+public:
+    void reportUnmapped(const softbus::core::plugin_system::DiscoveryRecord& record) override
+    {
+        softbus::message_bus::pipeline::DiscoverySample sample;
+        sample.compositeKey = record.compositeKey;
+        sample.endpointHash = record.endpointHash;
+        sample.protocol = record.protocol;
+        sample.slaveId = record.slaveId;
+        sample.registerAddress = record.registerAddress;
+        sample.rawValue = record.rawValue;
+        sample.lastSeenMs = record.lastSeenMs;
+        softbus::message_bus::pipeline::DiscoveryPool::instance().reportUnmapped(sample);
+    }
+};
+
 void endpointFramerThread(std::shared_ptr<EndpointFramingState> st)
 {
     while (st->running.load(std::memory_order_acquire)) {
@@ -515,7 +553,16 @@ std::vector<softbus::core::models::DOMMessage> PipelineEngine::stage2Parser(cons
         std::lock_guard<std::mutex> lk(m_mapperSessionMutex);
         auto it = m_mapperSessions.find(mapperKey);
         if (it == m_mapperSessions.end() || !it.value()) {
-            auto s = mapperPlugin->createMapperSession();
+            if (!m_mappingLookup) {
+                m_mappingLookup = std::make_unique<MappingLookupAdapter>();
+            }
+            if (!m_discoverySink) {
+                m_discoverySink = std::make_unique<DiscoverySinkAdapter>();
+            }
+            softbus::core::plugin_system::MapperSessionContext context;
+            context.mappingLookup = m_mappingLookup.get();
+            context.discoverySink = m_discoverySink.get();
+            auto s = mapperPlugin->createMapperSession(context);
             m_mapperSessions.insert(
                 mapperKey,
                 s ? std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>(std::move(s)) : nullptr);

src/message_bus/pipeline/PipelineEngine.h

@@ -19,7 +19,7 @@
 #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/rules/DynamicRuleRegistry.h"
 #include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h"
 
 namespace softbus::message_bus::pipeline
@@ -135,6 +135,8 @@ private:
     QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolSession>> m_parserSessions;
     std::mutex m_mapperSessionMutex;
     QHash<SessionKey, std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>> m_mapperSessions;
+    std::unique_ptr<softbus::core::plugin_system::IMappingLookup> m_mappingLookup;
+    std::unique_ptr<softbus::core::plugin_system::IDiscoverySink> m_discoverySink;
 
     std::mutex                    m_seqMutex;
     QHash<std::uint32_t, std::uint64_t> m_nextFrameSeqByEndpoint;

src/message_bus/pipeline/discovery/DiscoveryPool.cpp

@@ -0,0 +1,214 @@
+#include "message_bus/pipeline/discovery/DiscoveryPool.h"
+
+#include <algorithm>
+
+namespace softbus::message_bus::pipeline
+{
+
+DiscoveryPool& DiscoveryPool::instance()
+{
+    static DiscoveryPool pool;
+    return pool;
+}
+
+void DiscoveryPool::updateConfig(const Config& cfg)
+{
+    QMutexLocker lk(&m_mtx);
+    m_cfg = cfg;
+    if (m_cfg.maxEntries == 0) {
+        m_cfg.maxEntries = 1;
+    }
+    if (m_cfg.minSampleIntervalMs < 0) {
+        m_cfg.minSampleIntervalMs = 0;
+    }
+    if (m_cfg.sniffTtlMs < 0) {
+        m_cfg.sniffTtlMs = 0;
+    }
+    enforceCapacityLocked();
+}
+
+DiscoveryPool::Config DiscoveryPool::config() const
+{
+    QMutexLocker lk(&m_mtx);
+    return m_cfg;
+}
+
+void DiscoveryPool::startSniffing(std::int64_t ttlMs)
+{
+    const std::int64_t now = QDateTime::currentMSecsSinceEpoch();
+    QMutexLocker lk(&m_mtx);
+    m_sniffingEnabled.store(true, std::memory_order_release);
+    m_sniffStartedAtMs = now;
+    const std::int64_t effectiveTtl = ttlMs > 0 ? ttlMs : m_cfg.sniffTtlMs;
+    m_sniffExpiresAtMs = effectiveTtl > 0 ? (now + effectiveTtl) : 0;
+}
+
+void DiscoveryPool::stopSniffing()
+{
+    m_sniffingEnabled.store(false, std::memory_order_release);
+    QMutexLocker lk(&m_mtx);
+    m_sniffExpiresAtMs = 0;
+}
+
+bool DiscoveryPool::isSniffingEnabled() const
+{
+    return m_sniffingEnabled.load(std::memory_order_acquire);
+}
+
+void DiscoveryPool::tick()
+{
+    if (!isSniffingEnabled()) {
+        return;
+    }
+    const std::int64_t now = QDateTime::currentMSecsSinceEpoch();
+    QMutexLocker lk(&m_mtx);
+    if (m_sniffExpiresAtMs > 0 && now >= m_sniffExpiresAtMs) {
+        m_sniffingEnabled.store(false, std::memory_order_release);
+        m_sniffExpiresAtMs = 0;
+    }
+}
+
+void DiscoveryPool::reportUnmapped(const DiscoverySample& sample)
+{
+    if (!isSniffingEnabled()) {
+        return;
+    }
+    tick();
+    if (!isSniffingEnabled()) {
+        return;
+    }
+
+    QMutexLocker lk(&m_mtx);
+    auto it = m_slots.find(sample.compositeKey);
+    if (it == m_slots.end()) {
+        Slot s;
+        s.sample = sample;
+        s.sample.firstSeenMs = sample.lastSeenMs;
+        s.sample.hitCount = 1;
+        m_slots.emplace(sample.compositeKey, s);
+        enforceCapacityLocked();
+        return;
+    }
+
+    Slot& slot = it->second;
+    if (m_cfg.minSampleIntervalMs > 0 && sample.lastSeenMs > 0
+        && (sample.lastSeenMs - slot.sample.lastSeenMs) < m_cfg.minSampleIntervalMs) {
+        return;
+    }
+    slot.sample.rawValue = sample.rawValue;
+    slot.sample.lastSeenMs = sample.lastSeenMs;
+    slot.sample.endpointHash = sample.endpointHash;
+    slot.sample.protocol = sample.protocol;
+    slot.sample.slaveId = sample.slaveId;
+    slot.sample.registerAddress = sample.registerAddress;
+    ++slot.sample.hitCount;
+}
+
+void DiscoveryPool::clear()
+{
+    QMutexLocker lk(&m_mtx);
+    m_slots.clear();
+}
+
+QJsonObject DiscoveryPool::status() const
+{
+    QMutexLocker lk(&m_mtx);
+    QJsonObject o;
+    o.insert(QStringLiteral("sniffingEnabled"), m_sniffingEnabled.load(std::memory_order_acquire));
+    o.insert(QStringLiteral("sampleCount"), static_cast<qint64>(m_slots.size()));
+    o.insert(QStringLiteral("maxEntries"), static_cast<qint64>(m_cfg.maxEntries));
+    o.insert(QStringLiteral("minSampleIntervalMs"), static_cast<qint64>(m_cfg.minSampleIntervalMs));
+    o.insert(QStringLiteral("sniffStartedAtMs"), static_cast<qint64>(m_sniffStartedAtMs));
+    o.insert(QStringLiteral("sniffExpiresAtMs"), static_cast<qint64>(m_sniffExpiresAtMs));
+    return o;
+}
+
+QJsonArray DiscoveryPool::listSamples(std::size_t offset,
+                                      std::size_t limit,
+                                      std::uint32_t endpointFilter,
+                                      softbus::core::models::ProtocolType protocolFilter,
+                                      const std::unordered_map<std::uint64_t, QString>& mappedKeys) const
+{
+    QMutexLocker lk(&m_mtx);
+    std::vector<DiscoverySample> rows;
+    rows.reserve(m_slots.size());
+    for (const auto& kv : m_slots) {
+        const auto& sample = kv.second.sample;
+        if (endpointFilter != 0 && sample.endpointHash != endpointFilter) {
+            continue;
+        }
+        if (protocolFilter != softbus::core::models::ProtocolType::UNKNOWN && sample.protocol != protocolFilter) {
+            continue;
+        }
+        if (mappedKeys.find(sample.compositeKey) != mappedKeys.end()) {
+            continue;
+        }
+        rows.push_back(sample);
+    }
+
+    std::sort(rows.begin(), rows.end(), [](const DiscoverySample& a, const DiscoverySample& b) {
+        return a.lastSeenMs > b.lastSeenMs;
+    });
+
+    QJsonArray arr;
+    if (offset >= rows.size()) {
+        return arr;
+    }
+    const std::size_t end = std::min(rows.size(), offset + limit);
+    for (std::size_t i = offset; i < end; ++i) {
+        const auto& s = rows[i];
+        QJsonObject o;
+        o.insert(QStringLiteral("compositeKey"), QString::number(s.compositeKey));
+        o.insert(QStringLiteral("endpointHash"), static_cast<qint64>(s.endpointHash));
+        o.insert(QStringLiteral("protocol"), protocolToString(s.protocol));
+        o.insert(QStringLiteral("slaveId"), static_cast<int>(s.slaveId));
+        o.insert(QStringLiteral("registerAddress"), static_cast<int>(s.registerAddress));
+        o.insert(QStringLiteral("rawValue"), s.rawValue);
+        o.insert(QStringLiteral("firstSeenMs"), static_cast<qint64>(s.firstSeenMs));
+        o.insert(QStringLiteral("lastSeenMs"), static_cast<qint64>(s.lastSeenMs));
+        o.insert(QStringLiteral("hitCount"), static_cast<qint64>(s.hitCount));
+        arr.push_back(o);
+    }
+    return arr;
+}
+
+void DiscoveryPool::enforceCapacityLocked()
+{
+    if (m_slots.size() <= m_cfg.maxEntries) {
+        return;
+    }
+    while (m_slots.size() > m_cfg.maxEntries) {
+        auto victim = m_slots.end();
+        for (auto it = m_slots.begin(); it != m_slots.end(); ++it) {
+            if (victim == m_slots.end() || it->second.sample.lastSeenMs < victim->second.sample.lastSeenMs) {
+                victim = it;
+            }
+        }
+        if (victim == m_slots.end()) {
+            break;
+        }
+        m_slots.erase(victim);
+    }
+}
+
+QString DiscoveryPool::protocolToString(softbus::core::models::ProtocolType protocol)
+{
+    switch (protocol) {
+    case softbus::core::models::ProtocolType::MODBUS_RTU:
+        return QStringLiteral("MODBUS_RTU");
+    case softbus::core::models::ProtocolType::MODBUS_ASCII:
+        return QStringLiteral("MODBUS_ASCII");
+    case softbus::core::models::ProtocolType::MODBUS_TCP:
+        return QStringLiteral("MODBUS_TCP");
+    case softbus::core::models::ProtocolType::CAN_RAW:
+        return QStringLiteral("CAN_RAW");
+    case softbus::core::models::ProtocolType::CAN_OPEN:
+        return QStringLiteral("CAN_OPEN");
+    case softbus::core::models::ProtocolType::UDP_CUSTOM:
+        return QStringLiteral("UDP_CUSTOM");
+    default:
+        return QStringLiteral("UNKNOWN");
+    }
+}
+
+} // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/discovery/DiscoveryPool.h

@@ -0,0 +1,81 @@
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+#include <unordered_map>
+#include <vector>
+
+#include <QDateTime>
+#include <QJsonArray>
+#include <QJsonObject>
+#include <QMutex>
+#include <QString>
+
+#include "core/models/RawBusMessage.h"
+
+namespace softbus::message_bus::pipeline
+{
+
+struct DiscoverySample
+{
+    std::uint64_t compositeKey{0};
+    std::uint32_t endpointHash{0};
+    softbus::core::models::ProtocolType protocol{softbus::core::models::ProtocolType::UNKNOWN};
+    std::uint8_t slaveId{0};
+    std::uint16_t registerAddress{0};
+    double rawValue{0.0};
+    std::int64_t firstSeenMs{0};
+    std::int64_t lastSeenMs{0};
+    std::uint64_t hitCount{0};
+};
+
+class DiscoveryPool
+{
+public:
+    struct Config
+    {
+        std::size_t maxEntries{4096};
+        std::int64_t minSampleIntervalMs{200};
+        std::int64_t sniffTtlMs{0};
+    };
+
+    static DiscoveryPool& instance();
+
+    void updateConfig(const Config& cfg);
+    Config config() const;
+
+    void startSniffing(std::int64_t ttlMs = 0);
+    void stopSniffing();
+    bool isSniffingEnabled() const;
+    void tick();
+
+    void reportUnmapped(const DiscoverySample& sample);
+    void clear();
+
+    QJsonObject status() const;
+    QJsonArray listSamples(std::size_t offset,
+                           std::size_t limit,
+                           std::uint32_t endpointFilter,
+                           softbus::core::models::ProtocolType protocolFilter,
+                           const std::unordered_map<std::uint64_t, QString>& mappedKeys) const;
+
+private:
+    DiscoveryPool() = default;
+    void enforceCapacityLocked();
+    static QString protocolToString(softbus::core::models::ProtocolType protocol);
+
+private:
+    struct Slot
+    {
+        DiscoverySample sample;
+    };
+
+    mutable QMutex m_mtx;
+    std::unordered_map<std::uint64_t, Slot> m_slots;
+    Config m_cfg;
+    std::atomic<bool> m_sniffingEnabled{false};
+    std::int64_t m_sniffStartedAtMs{0};
+    std::int64_t m_sniffExpiresAtMs{0};
+};
+
+} // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/mapping/MappingRegistry.cpp

@@ -0,0 +1,72 @@
+#include "message_bus/pipeline/mapping/MappingRegistry.h"
+
+namespace softbus::message_bus::pipeline
+{
+
+MappingRegistry& MappingRegistry::instance()
+{
+    static MappingRegistry reg;
+    return reg;
+}
+
+bool MappingRegistry::upsert(const MappingRule& rule)
+{
+    if (rule.compositeKey == 0 || rule.domPath.isEmpty()) {
+        return false;
+    }
+    QWriteLocker lk(&m_lock);
+    m_rules[rule.compositeKey] = rule;
+    return true;
+}
+
+bool MappingRegistry::remove(std::uint64_t compositeKey)
+{
+    QWriteLocker lk(&m_lock);
+    return m_rules.erase(compositeKey) > 0;
+}
+
+std::optional<MappingRule> MappingRegistry::find(std::uint64_t compositeKey) const
+{
+    QReadLocker lk(&m_lock);
+    auto it = m_rules.find(compositeKey);
+    if (it == m_rules.end()) {
+        return std::nullopt;
+    }
+    return it->second;
+}
+
+std::unordered_map<std::uint64_t, QString> MappingRegistry::snapshotMappedKeys() const
+{
+    QReadLocker lk(&m_lock);
+    std::unordered_map<std::uint64_t, QString> out;
+    out.reserve(m_rules.size());
+    for (const auto& kv : m_rules) {
+        out.emplace(kv.first, kv.second.domPath);
+    }
+    return out;
+}
+
+QJsonArray MappingRegistry::list() const
+{
+    QReadLocker lk(&m_lock);
+    QJsonArray arr;
+    for (const auto& kv : m_rules) {
+        const auto& rule = kv.second;
+        QJsonObject o;
+        o.insert(QStringLiteral("compositeKey"), QString::number(rule.compositeKey));
+        o.insert(QStringLiteral("domPath"), rule.domPath);
+        o.insert(QStringLiteral("scale"), rule.scale);
+        o.insert(QStringLiteral("offset"), rule.offset);
+        o.insert(QStringLiteral("unit"), rule.unit);
+        arr.push_back(o);
+    }
+    return arr;
+}
+
+QJsonObject MappingRegistry::status() const
+{
+    QReadLocker lk(&m_lock);
+    return QJsonObject{{QStringLiteral("mappingRuleCount"), static_cast<qint64>(m_rules.size())}};
+}
+
+} // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/mapping/MappingRegistry.h

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <cstdint>
+#include <optional>
+#include <unordered_map>
+
+#include <QJsonArray>
+#include <QJsonObject>
+#include <QReadWriteLock>
+#include <QString>
+
+namespace softbus::message_bus::pipeline
+{
+
+struct MappingRule
+{
+    std::uint64_t compositeKey{0};
+    QString domPath;
+    double scale{1.0};
+    double offset{0.0};
+    QString unit;
+};
+
+class MappingRegistry
+{
+public:
+    static MappingRegistry& instance();
+
+    bool upsert(const MappingRule& rule);
+    bool remove(std::uint64_t compositeKey);
+    std::optional<MappingRule> find(std::uint64_t compositeKey) const;
+
+    std::unordered_map<std::uint64_t, QString> snapshotMappedKeys() const;
+    QJsonArray list() const;
+    QJsonObject status() const;
+
+private:
+    MappingRegistry() = default;
+
+private:
+    mutable QReadWriteLock m_lock;
+    std::unordered_map<std::uint64_t, MappingRule> m_rules;
+};
+
+} // namespace softbus::message_bus::pipeline

src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp

@@ -1,4 +1,4 @@
-#include "message_bus/pipeline/DynamicRuleRegistry.h"
+#include "message_bus/pipeline/rules/DynamicRuleRegistry.h"
 
 #include <mutex>
 #include <shared_mutex>
@@ -84,7 +84,6 @@ QJsonObject DynamicRuleRegistry::status() const
 bool DynamicRuleRegistry::matchScope(
     const RuleScope& scope, std::uint32_t endpointHash, int deviceId, const QString& stableKey) const
 {
-    // scope: global + endpointHash/device/stableKey selective match
     const bool endpointHashOk = (scope.endpointHash == 0) || (scope.endpointHash == endpointHash);
     const bool deviceOk = scope.deviceId < 0 || scope.deviceId == deviceId;
     const bool stableKeyOk = scope.stableKey.isEmpty() || scope.stableKey == stableKey;
@@ -92,4 +91,3 @@ bool DynamicRuleRegistry::matchScope(
 }
 
 } // namespace softbus::message_bus::pipeline
-

src/message_bus/pipeline/rules/DynamicRuleRegistry.h

@@ -60,4 +60,3 @@ private:
 };
 
 } // namespace softbus::message_bus::pipeline
-