flowzl/softbus_daemon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dommessage

Browse Source
This commit is contained in:
flower_linux
2026-04-28 21:56:56 +08:00
parent 7459cc0f36
commit 7280fca08d
24 changed files with 848 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
src/message_bus/pipeline/PipelineEngine.cpp
View File

@@ -7,7 +7,11 @@
#include <functional>
#include "core/models/MessageRoutingUtils.h"
#include "device_bus/DeviceBus.h"
#include "device_bus/tree/DeviceTreeModel.h"
#include "core/metadata/ProfileRegistry.h"
#include "message_bus/pipeline/PipelineEngineApi.h"
#include "message_bus/pipeline/derivation/DerivationEngine.h"
#include "message_bus/pipeline/discovery/DiscoveryPool.h"
#include "message_bus/pipeline/mapping/MappingRegistry.h"
#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
@@ -37,24 +41,113 @@ struct EndpointFramingState
namespace
{
QString protocolNormForProfile(softbus::core::models::ProtocolType p)
{
using softbus::core::models::ProtocolType;
switch (p) {
case ProtocolType::MODBUS_RTU:
return QStringLiteral("modbus_rtu");
case ProtocolType::MODBUS_ASCII:
return QStringLiteral("modbus_ascii");
case ProtocolType::MODBUS_TCP:
return QStringLiteral("modbus_tcp");
default:
return {};
}
}
class MappingLookupAdapter final : public softbus::core::plugin_system::IMappingLookup
{
public:
std::optional<softbus::core::plugin_system::MappingLookupResult> find(std::uint64_t compositeKey) const override
std::optional<softbus::core::plugin_system::MappingLookupResult> find(
const softbus::core::models::RawBusMessage& raw,
std::uint64_t compositeKey) const override
{
auto rule = softbus::message_bus::pipeline::MappingRegistry::instance().find(compositeKey);
if (!rule.has_value()) {
if (rule.has_value()) {
softbus::core::plugin_system::MappingLookupResult out;
out.domPath = rule->domPath;
out.scale = rule->scale;
out.offset = rule->offset;
out.unit = rule->unit;
out.metadataId = rule->metadataId;
out.pointId = rule->pointId.isEmpty() ? rule->domPath : rule->pointId;
out.sourceTag = QStringLiteral("mapping");
return out;
}
const QString proto = protocolNormForProfile(raw.protocol);
if (proto.isEmpty() || raw.protocol != softbus::core::models::ProtocolType::MODBUS_RTU) {
return std::nullopt;
}
const std::uint16_t regAddr = static_cast<std::uint16_t>(compositeKey & 0xFFFFu);
const auto hit = softbus::core::metadata::ProfileRegistry::instance().findModbusRegisterMapping(
raw.deviceId, proto, regAddr);
if (!hit.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;
out.domPath = hit->domPath;
out.scale = hit->scale;
out.offset = hit->offset;
out.unit = hit->unit;
out.metadataId = hit->metadataId;
out.pointId = hit->pointId.isEmpty() ? hit->domPath : hit->pointId;
out.sourceTag = QStringLiteral("profile");
return out;
}
};
void attachPesmRuntimeContext(const softbus::core::models::RawBusMessage& raw,
std::vector<softbus::core::models::DOMMessage>& msgs)
{
if (msgs.empty()) {
return;
}
auto tree = softbus::device_bus::DeviceBus::instance().deviceTreeModel();
std::optional<softbus::devices::DeviceInfo> devInfo;
if (raw.deviceId >= 0) {
devInfo = softbus::device_bus::DeviceBus::instance().findDeviceById(raw.deviceId);
}
const DeviceTreeNode* node = nullptr;
if (tree && devInfo.has_value()) {
const int idx = tree->findIndexByEndpoint(devInfo->endpoint);
if (idx >= 0 && idx < tree->nodes().size()) {
node = &tree->nodes().at(idx);
}
}
for (auto& msg : msgs) {
msg.attributes.insert(QStringLiteral("endpointHash"), static_cast<qint64>(raw.endpointHash));
if (raw.deviceId >= 0) {
msg.attributes.insert(QStringLiteral("runtimeDeviceId"), raw.deviceId);
}
if (devInfo.has_value()) {
msg.attributes.insert(QStringLiteral("deviceEndpoint"), devInfo->endpoint);
if (!devInfo->protocol.isEmpty()) {
msg.attributes.insert(QStringLiteral("deviceProtocol"), devInfo->protocol);
}
}
if (node) {
msg.attributes.insert(QStringLiteral("deviceNodeId"), node->id);
const QJsonValue av = node->params.value(QStringLiteral("assetId"));
if (av.isString() && !av.toString().isEmpty()) {
msg.attributes.insert(QStringLiteral("assetId"), av.toString());
}
const QJsonValue roleV = node->params.value(QStringLiteral("treeRole"));
if (roleV.isString() && !roleV.toString().isEmpty()) {
msg.attributes.insert(QStringLiteral("treeRole"), roleV.toString());
}
}
if (!msg.pointId.isEmpty()) {
msg.attributes.insert(QStringLiteral("pointId"), msg.pointId);
}
if (!msg.metadataId.isEmpty()) {
msg.attributes.insert(QStringLiteral("metadataId"), msg.metadataId);
}
}
}
class DiscoverySinkAdapter final : public softbus::core::plugin_system::IDiscoverySink
{
public:
@@ -584,6 +677,18 @@ bool PipelineEngine::runStages234(const softbus::core::models::RawBusMessage& ct
<< "quality=" << static_cast<int>(enriched.quality)
<< "alarm=" << enriched.alarm
<< "traceCount=" << static_cast<int>(enriched.traceLog.size());
std::vector<softbus::core::models::EnrichedMessage> derived;
softbus::message_bus::pipeline::derivation::DerivationEngine::instance().deriveFromEnriched(enriched, derived);
for (const auto& d : derived) {
const QString dv = std::visit(
[](const auto& x) { return QString::number(static_cast<double>(x), 'g', 12); },
d.engineeringValue);
LOG_INFO() << "PipelineEngine: derived message"
<< "sourceMessageId=" << d.sourceMessageId
<< "engineeringValue=" << dv
<< "traceCount=" << static_cast<int>(d.traceLog.size());
}
}
return true;
}
@@ -652,6 +757,9 @@ std::vector<softbus::core::models::DOMMessage> PipelineEngine::stage2Parser(cons
softbus::core::plugin_system::MapperSessionContext context;
context.mappingLookup = m_mappingLookup.get();
context.discoverySink = m_discoverySink.get();
context.sniffingEnabled = []() {
return softbus::message_bus::pipeline::DiscoveryPool::instance().isSniffingEnabled();
};
auto s = mapperPlugin->createMapperSession(context);
m_mapperSessions.insert(
mapperKey,
@@ -663,7 +771,9 @@ std::vector<softbus::core::models::DOMMessage> PipelineEngine::stage2Parser(cons
}
session = it.value();
}
return session->map(ctx, envelope);
auto messages = session->map(ctx, envelope);
attachPesmRuntimeContext(ctx, messages);
return messages;
}
bool PipelineEngine::stage3Filter(softbus::core::models::DOMMessage& message)

44
src/message_bus/pipeline/derivation/DerivationEngine.cpp Normal file
View File

@@ -0,0 +1,44 @@
#include "message_bus/pipeline/derivation/DerivationEngine.h"
#include <QJsonArray>
namespace softbus::message_bus::pipeline::derivation
{
DerivationEngine& DerivationEngine::instance()
{
static DerivationEngine eng;
return eng;
}
void DerivationEngine::deriveFromEnriched(const softbus::core::models::EnrichedMessage& in,
std::vector<softbus::core::models::EnrichedMessage>& outAppend) const
{
if (!m_enabled) {
return;
}
if (!in.attributes.value(QStringLiteral("sb.deriveMirror")).toBool()) {
return;
}
const QString srcPoint = in.attributes.value(QStringLiteral("pointId")).toString();
if (srcPoint.isEmpty()) {
return;
}
// MVP：若存在派生后缀配置则复制一份工程值到派生点（示例约定：pointId + ".mirror"）
const QString mirrorId = srcPoint + QStringLiteral(".mirror");
softbus::core::models::EnrichedMessage derived;
derived.sourceMessageId = mirrorId;
derived.logicalName = in.logicalName + QStringLiteral(" (derived)");
derived.unit = in.unit;
derived.engineeringValue = in.engineeringValue;
derived.quality = in.quality;
derived.attributes = in.attributes;
derived.traceLog = in.traceLog;
derived.attributes.insert(QStringLiteral("pointId"), mirrorId);
derived.attributes.insert(QStringLiteral("derivedFrom"), QJsonArray{srcPoint});
derived.attributes.insert(QStringLiteral("derivationRuleId"), QStringLiteral("builtin.mirror"));
derived.attributes.insert(QStringLiteral("derivationVersion"), QStringLiteral("1"));
outAppend.push_back(std::move(derived));
}
} // namespace softbus::message_bus::pipeline::derivation

30
src/message_bus/pipeline/derivation/DerivationEngine.h Normal file
View File

@@ -0,0 +1,30 @@
#pragma once
#include <vector>
#include "core/models/EnrichedMessage.h"
namespace softbus::message_bus::pipeline::derivation
{
/// 最小推导层：基于已富化的标准点生成可追溯派生点（占位 + 单规则 MVP）
class DerivationEngine
{
public:
static DerivationEngine& instance();
/// 根据当前配置，从输入生成零条或多条派生富化消息（不改变输入）
void deriveFromEnriched(const softbus::core::models::EnrichedMessage& in,
std::vector<softbus::core::models::EnrichedMessage>& outAppend) const;
void setEnabled(bool on) { m_enabled = on; }
bool enabled() const { return m_enabled; }
private:
DerivationEngine() = default;
private:
bool m_enabled{true};
};
} // namespace softbus::message_bus::pipeline::derivation

141
src/message_bus/pipeline/mapping/MappingRegistry.cpp
View File

@@ -1,5 +1,13 @@
#include "message_bus/pipeline/mapping/MappingRegistry.h"
#include <QDateTime>
#include <QFile>
#include <QJsonArray>
#include <QJsonDocument>
#include <QJsonObject>
#include <QJsonParseError>
#include <QSaveFile>
namespace softbus::message_bus::pipeline
{
@@ -15,14 +23,23 @@ bool MappingRegistry::upsert(const MappingRule& rule)
return false;
}
QWriteLocker lk(&m_lock);
m_rules[rule.compositeKey] = rule;
MappingRule r = rule;
if (r.updatedAtMs <= 0) {
r.updatedAtMs = QDateTime::currentMSecsSinceEpoch();
}
m_rules[rule.compositeKey] = std::move(r);
++m_catalogVersion;
return true;
}
bool MappingRegistry::remove(std::uint64_t compositeKey)
{
QWriteLocker lk(&m_lock);
return m_rules.erase(compositeKey) > 0;
const bool ok = m_rules.erase(compositeKey) > 0;
if (ok) {
++m_catalogVersion;
}
return ok;
}
std::optional<MappingRule> MappingRegistry::find(std::uint64_t compositeKey) const
@@ -58,6 +75,21 @@ QJsonArray MappingRegistry::list() const
o.insert(QStringLiteral("scale"), rule.scale);
o.insert(QStringLiteral("offset"), rule.offset);
o.insert(QStringLiteral("unit"), rule.unit);
if (!rule.metadataId.isEmpty()) {
o.insert(QStringLiteral("metadataId"), rule.metadataId);
}
if (!rule.pointId.isEmpty()) {
o.insert(QStringLiteral("pointId"), rule.pointId);
}
if (!rule.catalogVersion.isEmpty()) {
o.insert(QStringLiteral("catalogVersion"), rule.catalogVersion);
}
if (rule.updatedAtMs > 0) {
o.insert(QStringLiteral("updatedAtMs"), rule.updatedAtMs);
}
if (!rule.updatedBy.isEmpty()) {
o.insert(QStringLiteral("updatedBy"), rule.updatedBy);
}
arr.push_back(o);
}
return arr;
@@ -66,7 +98,110 @@ QJsonArray MappingRegistry::list() const
QJsonObject MappingRegistry::status() const
{
QReadLocker lk(&m_lock);
return QJsonObject{{QStringLiteral("mappingRuleCount"), static_cast<qint64>(m_rules.size())}};
QJsonObject o;
o.insert(QStringLiteral("mappingRuleCount"), static_cast<qint64>(m_rules.size()));
o.insert(QStringLiteral("mappingCatalogVersion"), m_catalogVersion);
return o;
}
bool MappingRegistry::loadPersistedFile(const QString& path)
{
QFile f(path);
if (!f.exists()) {
return true;
}
if (!f.open(QIODevice::ReadOnly | QIODevice::Text)) {
return false;
}
QJsonParseError pe;
const QJsonDocument doc = QJsonDocument::fromJson(f.readAll(), &pe);
if (pe.error != QJsonParseError::NoError || !doc.isObject()) {
return false;
}
const QJsonObject root = doc.object();
const qint64 ver = root.value(QStringLiteral("catalogVersion")).toVariant().toLongLong();
const QJsonArray items = root.value(QStringLiteral("items")).toArray();
QWriteLocker lk(&m_lock);
m_rules.clear();
for (const auto& v : items) {
if (!v.isObject()) {
continue;
}
const QJsonObject o = v.toObject();
MappingRule rule;
rule.compositeKey = o.value(QStringLiteral("compositeKey")).toString().toULongLong();
rule.domPath = o.value(QStringLiteral("domPath")).toString();
rule.scale = o.value(QStringLiteral("scale")).toDouble(1.0);
rule.offset = o.value(QStringLiteral("offset")).toDouble(0.0);
rule.unit = o.value(QStringLiteral("unit")).toString();
rule.metadataId = o.value(QStringLiteral("metadataId")).toString();
rule.pointId = o.value(QStringLiteral("pointId")).toString();
rule.catalogVersion = o.value(QStringLiteral("catalogVersion")).toString();
rule.updatedAtMs = o.value(QStringLiteral("updatedAtMs")).toVariant().toLongLong();
rule.updatedBy = o.value(QStringLiteral("updatedBy")).toString();
if (rule.compositeKey != 0 && !rule.domPath.isEmpty()) {
m_rules[rule.compositeKey] = rule;
}
}
if (ver > 0) {
m_catalogVersion = ver;
}
return true;
}
bool MappingRegistry::savePersistedFile(const QString& path) const
{
QJsonArray items;
{
QReadLocker lk(&m_lock);
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);
if (!rule.metadataId.isEmpty()) {
o.insert(QStringLiteral("metadataId"), rule.metadataId);
}
if (!rule.pointId.isEmpty()) {
o.insert(QStringLiteral("pointId"), rule.pointId);
}
if (!rule.catalogVersion.isEmpty()) {
o.insert(QStringLiteral("catalogVersion"), rule.catalogVersion);
}
if (rule.updatedAtMs > 0) {
o.insert(QStringLiteral("updatedAtMs"), rule.updatedAtMs);
}
if (!rule.updatedBy.isEmpty()) {
o.insert(QStringLiteral("updatedBy"), rule.updatedBy);
}
items.push_back(o);
}
}
qint64 verSnap = 1;
{
QReadLocker lk(&m_lock);
verSnap = m_catalogVersion;
}
QJsonObject root;
root.insert(QStringLiteral("catalogVersion"), verSnap);
root.insert(QStringLiteral("items"), items);
QSaveFile out(path);
if (!out.open(QIODevice::WriteOnly | QIODevice::Text)) {
return false;
}
const QByteArray bytes = QJsonDocument(root).toJson(QJsonDocument::Indented);
if (out.write(bytes) != bytes.size()) {
out.cancelWriting();
return false;
}
return out.commit();
}
} // namespace softbus::message_bus::pipeline

10
src/message_bus/pipeline/mapping/MappingRegistry.h
View File

@@ -19,6 +19,11 @@ struct MappingRule
double scale{1.0};
double offset{0.0};
QString unit;
QString metadataId;
QString pointId;
QString catalogVersion;
qint64 updatedAtMs{0};
QString updatedBy;
};
class MappingRegistry
@@ -34,12 +39,17 @@ public:
QJsonArray list() const;
QJsonObject status() const;
bool loadPersistedFile(const QString& path);
bool savePersistedFile(const QString& path) const;
qint64 mappingCatalogVersion() const { return m_catalogVersion; }
private:
MappingRegistry() = default;
private:
mutable QReadWriteLock m_lock;
std::unordered_map<std::uint64_t, MappingRule> m_rules;
qint64 m_catalogVersion{1};
};
} // namespace softbus::message_bus::pipeline

32
src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp
View File

@@ -13,14 +13,18 @@ bool ProtocolParseFilter::process(softbus::core::models::DOMMessage& message)
if (!ensureLoaded()) {
return true;
}
// Message-stage filter: only normalize quality by metadata range.
if (!message.metaRef) {
return true;
const softbus::core::models::MetadataDef* meta = message.metaRef;
if (!meta && !message.metadataId.isEmpty()) {
meta = m_metadataRegistry.findById(message.metadataId);
}
const auto* meta = m_metadataRegistry.findById(message.metaRef->metadataId);
if (!meta) {
return true;
}
if (message.metadataId.isEmpty()) {
message.metadataId = meta->metadataId;
}
message.attributes.insert(QStringLiteral("domain"),
softbus::core::models::dataDomainToString(meta->domain));
const double v = std::visit(
[](const auto& x) { return static_cast<double>(x); },
message.value);
@@ -36,12 +40,27 @@ 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.metadataId = message.metadataId;
out.pointId = message.pointId.isEmpty() ? message.messageId : message.pointId;
const softbus::core::models::MetadataDef* meta = message.metaRef;
if (!meta && !message.metadataId.isEmpty()) {
meta = m_metadataRegistry.findById(message.metadataId);
}
out.logicalName = meta ? meta->name : message.messageId;
out.unit = meta ? meta->unit : QString();
if (!meta && message.attributes.contains(QStringLiteral("unit"))) {
out.unit = message.attributes.value(QStringLiteral("unit")).toString();
}
out.engineeringValue = message.value;
out.quality = message.quality;
out.attributes = message.attributes;
out.traceLog = message.traceLog;
if (meta && !meta->entryCatalogVersion.isEmpty()) {
out.catalogVersion = meta->entryCatalogVersion;
} else if (!m_metadataRegistry.catalogVersion().isEmpty()) {
out.catalogVersion = m_metadataRegistry.catalogVersion();
}
return out;
}
@@ -64,4 +83,3 @@ bool ProtocolParseFilter::ensureLoaded() const
}
} // namespace softbus::message_bus::pipeline::stages::filters