diff --git a/CMakeLists.txt b/CMakeLists.txt index 66442e3..9b3d768 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -130,6 +130,7 @@ qt_add_executable(softbus_daemon src/core/plugin_system/IProtocolPlugin.h src/core/plugin_system/IProtocolFramerPlugin.h src/core/plugin_system/IProtocolMapperPlugin.h + src/core/plugin_system/IMapperDependencies.h src/core/plugin_system/PluginManager.h src/core/plugin_system/PluginManager.cpp src/core/plugin_system/PluginHost.h @@ -151,8 +152,12 @@ qt_add_executable(softbus_daemon # message bus pipeline src/core/models/PayloadRef.h - src/message_bus/pipeline/DynamicRuleRegistry.h - src/message_bus/pipeline/DynamicRuleRegistry.cpp + src/message_bus/pipeline/rules/DynamicRuleRegistry.h + src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp + src/message_bus/pipeline/discovery/DiscoveryPool.h + src/message_bus/pipeline/discovery/DiscoveryPool.cpp + src/message_bus/pipeline/mapping/MappingRegistry.h + src/message_bus/pipeline/mapping/MappingRegistry.cpp src/message_bus/pipeline/PipelineEngine.h src/message_bus/pipeline/PipelineEngine.cpp src/message_bus/pipeline/protocol/ProtocolEnvelope.h @@ -199,6 +204,12 @@ target_link_libraries(softbus_daemon Qt6::HttpServer ) +# Export daemon symbols so runtime-loaded protocol plugins can resolve +# shared singleton APIs (e.g. DiscoveryPool/MappingRegistry) from host process. +# if(UNIX AND NOT APPLE) +# target_link_options(softbus_daemon PRIVATE "-Wl,--export-dynamic") +# endif() + # Platform-specific deps if(UNIX AND NOT APPLE) find_package(PkgConfig QUIET) diff --git a/bin/softbus_daemon b/bin/softbus_daemon index f1609a4..666a12d 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/docs/数据流转.md b/docs/数据流转.md index 798cc9a..e789fb8 100644 --- a/docs/数据流转.md +++ b/docs/数据流转.md @@ -152,4 +152,55 @@ SerialQtDriver(read callback) 1. `traceId` 建议在全链路透传到最终发布侧,便于定位帧级问题。 2. 为 Stage2/Stage3/Stage4 增加细粒度指标(成功率、耗时、drop 原因)。 -3. 文档后续可新增“协议插件开发约定”章节(framer/parser/mapper 三类插件接口对照)。 \ No newline at end of file +3. 文档后续可新增“协议插件开发约定”章节(framer/parser/mapper 三类插件接口对照)。 + +--- + +## 8. 无规则冷启动与发现池(新增) + +### 8.1 设计目标 + +- 默认保持极致性能:**未命中映射规则的寄存器不会产出 DOMMessage**。 +- 仅在配置页面开启时进入“嗅探模式(Sniffer Mode)”,将未命中点位以低成本写入内存发现池。 +- 发现池只维护“最近值”,不做全量历史存储,避免高频总线拖垮系统。 + +### 8.2 Mapper 行为(Modbus RTU) + +`ModbusRtuProtocolMapperPlugin` 当前行为: + +1. 对 `ReadHoldingRegistersRequest` 记录 `unitId -> startAddress`,用于后续响应地址回填。 +2. 对 `ReadHoldingRegistersResponse` 遍历寄存器并构造 `compositeKey`: + - `compositeKey = endpointHash + protocol + slaveId + registerAddress`(位拼接)。 +3. 若 `MappingRegistry` 命中: + - 生成 `DOMMessage`,`messageId = domPath`,并应用 `scale/offset/unit`。 +4. 若未命中: + - 仅在 `DiscoveryPool.sniffingEnabled=true` 时写入发现池样本; + - 主链路继续 `continue`,不生成业务消息。 + +### 8.3 DiscoveryPool 约束 + +- 数据结构:`unordered_map`。 +- 样本内容:`compositeKey/endpointHash/protocol/slaveId/registerAddress/rawValue/firstSeenMs/lastSeenMs/hitCount`。 +- 性能保护: + - `maxEntries` 限制最大条目数(超限按最旧 `lastSeenMs` 淘汰); + - `minSampleIntervalMs` 做按键限频,降低写放大; + - 支持 `ttlMs` 自动关闭嗅探,防止遗留长期开启。 + +### 8.4 REST API(配置工作台) + +- `POST /api/v1/discovery/start`:开启嗅探(可传 `ttlMs`)。 +- `POST /api/v1/discovery/stop`:关闭嗅探。 +- `GET /api/v1/discovery/status`:查看嗅探状态与池容量信息。 +- `GET /api/v1/discovery/samples`:拉取发现样本(支持 `offset/limit/endpointHash/protocol`)。 +- `POST /api/v1/discovery/clear`:清空发现池。 +- `POST /api/v1/mapping/bind`:新增或更新绑定规则(`compositeKey -> domPath + scale/offset/unit`)。 +- `POST /api/v1/mapping/unbind`:删除绑定规则。 +- `GET /api/v1/mapping/list`:查看当前绑定规则。 + +### 8.5 双向匹配工作流 + +1. Vue 打开配置页 -> 调用 `discovery/start`。 +2. 左侧“发现列表”轮询 `discovery/samples`。 +3. 用户执行 Bottom-Up(从样本创建 DOM 绑定)或 Top-Down(将样本拖到 DOM 占位符)。 +4. 前端调用 `mapping/bind`,规则即时生效。 +5. 下一帧同 `compositeKey` 数据在 mapper 命中,进入 DOM 主链路;该点位在发现列表中自动消失(查询时过滤已映射键)。 \ No newline at end of file diff --git a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp index d17a258..c8485ad 100644 --- a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp +++ b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp @@ -1,6 +1,8 @@ #include "plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h" #include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" +#include +#include namespace softbus::message_bus::pipeline::protocol::modbusrtu { @@ -11,6 +13,22 @@ namespace class ModbusRtuMapperSession final : public softbus::core::plugin_system::IProtocolMapperSession { public: + explicit ModbusRtuMapperSession(softbus::core::plugin_system::MapperSessionContext context) + : m_context(std::move(context)) + { + } + + static std::uint64_t buildCompositeKey(std::uint32_t endpointHash, + softbus::core::models::ProtocolType protocol, + std::uint8_t slaveId, + std::uint16_t address) + { + return (static_cast(endpointHash) << 32) + | (static_cast(static_cast(protocol)) << 24) + | (static_cast(slaveId) << 16) + | static_cast(address); + } + std::vector map( const softbus::core::models::RawBusMessage& rawContext, const softbus::message_bus::pipeline::protocol::ProtocolEnvelope& envelope) override @@ -27,17 +45,39 @@ public: if (const auto* resp = std::get_if(pdu)) { out.reserve(static_cast(resp->registers.size())); + const std::uint16_t baseAddress = m_lastReadHoldingStartAddress.value(resp->unitId, 0); for (int i = 0; i < resp->registers.size(); ++i) { + const auto addr = static_cast(baseAddress + static_cast(i)); + const auto compKey = buildCompositeKey( + rawContext.endpointHash, rawContext.protocol, resp->unitId, addr); + const auto mapRule = m_context.mappingLookup ? m_context.mappingLookup->find(compKey) : std::nullopt; + const double rawValue = static_cast(resp->registers[static_cast(i)]); + if (!mapRule.has_value()) { + if (m_context.discoverySink) { + softbus::core::plugin_system::DiscoveryRecord rec; + rec.compositeKey = compKey; + rec.endpointHash = rawContext.endpointHash; + rec.protocol = rawContext.protocol; + rec.slaveId = resp->unitId; + rec.registerAddress = addr; + rec.rawValue = rawValue; + rec.lastSeenMs = rawContext.timestampMs; + m_context.discoverySink->reportUnmapped(rec); + } + continue; + } softbus::core::models::DOMMessage msg; - msg.messageId = QStringLiteral("%1.reg.%2") - .arg(rawContext.routingKey == 0 - ? QString::number(rawContext.deviceId) - : QString::number(rawContext.routingKey)) - .arg(i); + msg.messageId = mapRule->domPath; msg.timestamp = rawContext.timestampMs; - msg.value = static_cast(resp->registers[static_cast(i)]); + msg.value = (rawValue * mapRule->scale) + mapRule->offset; msg.quality = softbus::core::models::DataQuality::GOOD; msg.attributes.insert(QStringLiteral("sourceEndpoint"), QString::number(rawContext.endpointHash)); + msg.attributes.insert(QStringLiteral("compositeKey"), QString::number(compKey)); + msg.attributes.insert(QStringLiteral("slaveId"), static_cast(resp->unitId)); + msg.attributes.insert(QStringLiteral("registerAddress"), static_cast(addr)); + if (!mapRule->unit.isEmpty()) { + msg.attributes.insert(QStringLiteral("unit"), mapRule->unit); + } out.push_back(std::move(msg)); } return out; @@ -45,24 +85,16 @@ public: if (const auto* req = std::get_if(pdu)) { - softbus::core::models::DOMMessage msg; - msg.messageId = QStringLiteral("%1.req.read_holding") - .arg(rawContext.routingKey == 0 - ? QString::number(rawContext.deviceId) - : QString::number(rawContext.routingKey)); - msg.timestamp = rawContext.timestampMs; - msg.value = static_cast(req->quantity); - msg.quality = softbus::core::models::DataQuality::GOOD; - msg.attributes.insert(QStringLiteral("sourceEndpoint"), QString::number(rawContext.endpointHash)); - msg.attributes.insert(QStringLiteral("frameType"), QStringLiteral("request")); - msg.attributes.insert(QStringLiteral("startAddress"), static_cast(req->startAddress)); - msg.attributes.insert(QStringLiteral("quantity"), static_cast(req->quantity)); - out.push_back(std::move(msg)); + m_lastReadHoldingStartAddress.insert(req->unitId, req->startAddress); } return out; } void reset() override {} + +private: + softbus::core::plugin_system::MapperSessionContext m_context; + QHash m_lastReadHoldingStartAddress; }; class ModbusRtuProtocolMapperPlugin final : public softbus::core::plugin_system::IProtocolMapperPlugin @@ -75,7 +107,12 @@ public: } std::unique_ptr createMapperSession() override { - return std::make_unique(); + return std::make_unique(softbus::core::plugin_system::MapperSessionContext{}); + } + std::unique_ptr createMapperSession( + const softbus::core::plugin_system::MapperSessionContext& ctx) override + { + return std::make_unique(ctx); } }; diff --git a/src/api/API_INTERFACE_TABLE.md b/src/api/API_INTERFACE_TABLE.md index c8d6289..a6de5e9 100644 --- a/src/api/API_INTERFACE_TABLE.md +++ b/src/api/API_INTERFACE_TABLE.md @@ -1,8 +1,9 @@ # softbus_daemon API 接口表 -本文档汇总 `src/api/ipc_dbus` 当前对外接口,分为: +本文档汇总 `src/api` 当前对外接口,分为: - D-Bus 直接方法 - `executeCommand` JSON 命令接口 +- HTTP REST 接口 ## 1) D-Bus 服务基础信息 @@ -53,15 +54,49 @@ | action | params 关键字段 | 成功 code | 失败 code | 说明 | |---|---|---:|---:|---| -| `add_filter` | `id`, `endpoint?`, `deviceId?`, `stableKey?`, `condition?` | `0` | `4001` | 添加过滤规则 | +| `add_filter` | `id`, `endpointHash?`, `deviceId?`, `stableKey?`, `condition?` | `0` | `4001` | 添加过滤规则 | | `remove_filter` | `id` | `0` | `4041` | 删除过滤规则 | -| `add_transform_rule` | `id`, `endpoint?`, `deviceId?`, `stableKey?`, `scale?`, `offset?`, `unit?` | `0` | `4002` | 添加转换规则 | +| `add_transform_rule` | `id`, `endpointHash?`, `deviceId?`, `stableKey?`, `scale?`, `offset?`, `unit?` | `0` | `4002` | 添加转换规则 | | `remove_transform_rule` | `id` | `0` | `4042` | 删除转换规则 | +| `add_mapping_rule` | `compositeKey`, `domPath`, `scale?`, `offset?`, `unit?` | `0` | `4004` | 新增/更新映射规则 | +| `remove_mapping_rule` | `compositeKey` | `0` | `4043` | 删除映射规则 | +| `list_mapping_rules` | 无 | `0` | - | 返回映射规则列表 | +| `discovery_start` | `ttlMs?` | `0` | - | 开启嗅探模式 | +| `discovery_stop` | 无 | `0` | - | 关闭嗅探模式 | +| `discovery_clear` | 无 | `0` | - | 清空发现池 | | `list_rules` | 无 | `0` | - | 返回规则计数状态 | -| `get_status` | 无 | `0` | - | 返回规则计数状态 | +| `get_status` | 无 | `0` | - | 返回规则计数 + discovery/mapping 状态 | | 其他未知 action | - | - | `4040` | 未知动作 | -## 5) 通用错误码约定(当前实现) +## 5) HTTP REST 路由表(RestApiRoutes) + +### 5.1 通用路由 + +| 方法 | 路径 | 说明 | +|---|---|---| +| `GET` | `/health` | 健康检查,返回 `{"ok": true}` | +| `GET` | `/v1/status` | 透传 `get_status` 的状态响应 | +| `POST` | `/v1/command` | 统一命令入口(请求体为 executeCommand JSON) | + +### 5.2 Discovery 路由 + +| 方法 | 路径 | 请求参数 | 说明 | +|---|---|---|---| +| `POST` | `/api/v1/discovery/start` | body: `ttlMs?` | 开启嗅探;可设置自动关闭 TTL | +| `POST` | `/api/v1/discovery/stop` | 无 | 关闭嗅探 | +| `GET` | `/api/v1/discovery/status` | 无 | 获取发现池运行状态 | +| `POST` | `/api/v1/discovery/clear` | 无 | 清空发现池样本 | +| `GET` | `/api/v1/discovery/samples` | query: `offset?`, `limit?`, `endpointHash?`, `protocol?` | 拉取样本(按更新时间倒序,且过滤已映射 key) | + +### 5.3 Mapping 路由 + +| 方法 | 路径 | 请求参数 | 说明 | +|---|---|---|---| +| `POST` | `/api/v1/mapping/bind` | body: `compositeKey`, `domPath`, `scale?`, `offset?`, `unit?` | 绑定或更新映射规则 | +| `POST` | `/api/v1/mapping/unbind` | body: `compositeKey` | 删除映射规则 | +| `GET` | `/api/v1/mapping/list` | 无 | 列出当前映射规则 | + +## 6) 通用错误码约定(当前实现) | code | message | 说明 | |---:|---|---| @@ -70,14 +105,19 @@ | `4001` | `invalid_filter_rule` | 过滤规则参数非法 | | `4002` | `invalid_transform_rule` | 转换规则参数非法 | | `4003` | `invalid_action` | action 为空或分发器不可用 | +| `4004` | `invalid_mapping_rule` | 映射规则参数非法 | | `4040` | `unknown_action` | action 未注册 | | `4041` | `filter_not_found` | 删除过滤规则时未找到 | | `4042` | `transform_rule_not_found` | 删除转换规则时未找到 | +| `4043` | `mapping_rule_not_found` | 删除映射规则时未找到 | -## 6) 维护建议 +## 7) 维护建议 - 新增 action 时,同时更新: - `CommandDispatcher::dispatch(...)` - 本文档第 4 节 action 表 +- 新增 REST 路由时,同时更新: + - `RestApiRoutes.cpp` + - 本文档第 5 节路由表 - 若扩展返回结构,保持 `code/message/data` 三元组不变,避免客户端兼容性问题。 diff --git a/src/api/http/RestApiRoutes.cpp b/src/api/http/RestApiRoutes.cpp index 1511e90..6474509 100644 --- a/src/api/http/RestApiRoutes.cpp +++ b/src/api/http/RestApiRoutes.cpp @@ -1,18 +1,164 @@ #include "api/http/RestApiRoutes.h" +#include #include #include #include #include +#include +#include #include +#include +#include #include #include "api/http/RestApiCommandHttp.h" #include "api/ipc_dbus/CommandDispatcher.h" +#include "message_bus/pipeline/discovery/DiscoveryPool.h" +#include "message_bus/pipeline/mapping/MappingRegistry.h" namespace softbus::api::http { +namespace +{ + +softbus::core::models::ProtocolType protocolFromString(const QString& s) +{ + const QString x = s.trimmed().toUpper(); + if (x == QStringLiteral("MODBUS_RTU")) { + return softbus::core::models::ProtocolType::MODBUS_RTU; + } + if (x == QStringLiteral("MODBUS_ASCII")) { + return softbus::core::models::ProtocolType::MODBUS_ASCII; + } + if (x == QStringLiteral("MODBUS_TCP")) { + return softbus::core::models::ProtocolType::MODBUS_TCP; + } + if (x == QStringLiteral("CAN_RAW")) { + return softbus::core::models::ProtocolType::CAN_RAW; + } + if (x == QStringLiteral("CAN_OPEN")) { + return softbus::core::models::ProtocolType::CAN_OPEN; + } + if (x == QStringLiteral("UDP_CUSTOM")) { + return softbus::core::models::ProtocolType::UDP_CUSTOM; + } + return softbus::core::models::ProtocolType::UNKNOWN; +} + +QJsonObject parseJsonBody(const QHttpServerRequest& request) +{ + QJsonParseError err; + const auto doc = QJsonDocument::fromJson(request.body(), &err); + if (err.error != QJsonParseError::NoError || !doc.isObject()) { + return {}; + } + return doc.object(); +} + +void registerDiscoveryRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/api/v1/discovery/start"), QHttpServerRequest::Method::Post, routeContext, + [](const QHttpServerRequest& request) { + auto& pool = softbus::message_bus::pipeline::DiscoveryPool::instance(); + const auto payload = parseJsonBody(request); + const std::int64_t ttlMs = payload.value(QStringLiteral("ttlMs")).toInteger(0); + pool.startSniffing(ttlMs); + return QHttpServerResponse(pool.status(), QHttpServerResponse::StatusCode::Ok); + }); + + server->route(QStringLiteral("/api/v1/discovery/stop"), QHttpServerRequest::Method::Post, routeContext, []() { + auto& pool = softbus::message_bus::pipeline::DiscoveryPool::instance(); + pool.stopSniffing(); + return QHttpServerResponse(pool.status(), QHttpServerResponse::StatusCode::Ok); + }); + + server->route(QStringLiteral("/api/v1/discovery/status"), QHttpServerRequest::Method::Get, routeContext, []() { + auto& pool = softbus::message_bus::pipeline::DiscoveryPool::instance(); + pool.tick(); + return QHttpServerResponse(pool.status(), QHttpServerResponse::StatusCode::Ok); + }); + + server->route(QStringLiteral("/api/v1/discovery/clear"), QHttpServerRequest::Method::Post, routeContext, []() { + auto& pool = softbus::message_bus::pipeline::DiscoveryPool::instance(); + pool.clear(); + return QHttpServerResponse(pool.status(), QHttpServerResponse::StatusCode::Ok); + }); + + server->route(QStringLiteral("/api/v1/discovery/samples"), QHttpServerRequest::Method::Get, routeContext, + [](const QHttpServerRequest& request) { + auto& pool = softbus::message_bus::pipeline::DiscoveryPool::instance(); + auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); + pool.tick(); + const QUrlQuery query(request.url().query()); + const std::size_t offset = + static_cast(std::max(0, query.queryItemValue(QStringLiteral("offset")).toInt(0))); + bool limitOk = false; + const int limitCandidate = query.queryItemValue(QStringLiteral("limit")).toInt(&limitOk, 10); + const int limitRaw = limitOk ? limitCandidate : 200; + const std::size_t limit = static_cast(std::clamp(limitRaw, 1, 500)); + const std::uint32_t endpointHash = static_cast( + query.queryItemValue(QStringLiteral("endpointHash")).toUInt(nullptr, 10)); + const auto protocol = protocolFromString(query.queryItemValue(QStringLiteral("protocol"))); + + QJsonObject o; + o.insert(QStringLiteral("items"), + pool.listSamples(offset, limit, endpointHash, protocol, mappingReg.snapshotMappedKeys())); + o.insert(QStringLiteral("offset"), static_cast(offset)); + o.insert(QStringLiteral("limit"), static_cast(limit)); + o.insert(QStringLiteral("status"), pool.status()); + return QHttpServerResponse(o, QHttpServerResponse::StatusCode::Ok); + }); +} + +void registerMappingRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/api/v1/mapping/bind"), QHttpServerRequest::Method::Post, routeContext, + [](const QHttpServerRequest& request) { + const auto payload = parseJsonBody(request); + softbus::message_bus::pipeline::MappingRule rule; + rule.compositeKey = payload.value(QStringLiteral("compositeKey")).toString().toULongLong(); + rule.domPath = payload.value(QStringLiteral("domPath")).toString(); + rule.scale = payload.value(QStringLiteral("scale")).toDouble(1.0); + rule.offset = payload.value(QStringLiteral("offset")).toDouble(0.0); + rule.unit = payload.value(QStringLiteral("unit")).toString(); + + auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); + if (!mappingReg.upsert(rule)) { + return QHttpServerResponse( + QJsonObject{{QStringLiteral("error"), QStringLiteral("invalid_mapping_rule")}}, + QHttpServerResponse::StatusCode::BadRequest); + } + return QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), true}, + {QStringLiteral("compositeKey"), + QString::number(rule.compositeKey)}}, + QHttpServerResponse::StatusCode::Ok); + }); + + server->route(QStringLiteral("/api/v1/mapping/unbind"), QHttpServerRequest::Method::Post, routeContext, + [](const QHttpServerRequest& request) { + const auto payload = parseJsonBody(request); + const std::uint64_t compositeKey = + payload.value(QStringLiteral("compositeKey")).toString().toULongLong(); + auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); + const bool ok = mappingReg.remove(compositeKey); + return QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), ok}, + {QStringLiteral("compositeKey"), + QString::number(compositeKey)}}, + ok ? QHttpServerResponse::StatusCode::Ok + : QHttpServerResponse::StatusCode::NotFound); + }); + + server->route(QStringLiteral("/api/v1/mapping/list"), QHttpServerRequest::Method::Get, routeContext, []() { + auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); + return QHttpServerResponse(QJsonObject{{QStringLiteral("items"), mappingReg.list()}}, + QHttpServerResponse::StatusCode::Ok); + }); +} + +} // namespace + void registerSoftbusRestApiRoutes(QHttpServer* server, QObject* routeContext, softbus::api::ipc::CommandDispatcher* dispatcher) @@ -33,6 +179,8 @@ void registerSoftbusRestApiRoutes(QHttpServer* server, [dispatcher](const QHttpServerRequest& request) { return handleExecuteCommandHttpBody(request.body(), *dispatcher); }); + registerDiscoveryRoutes(server, routeContext); + registerMappingRoutes(server, routeContext); server->setMissingHandler(routeContext, [](const QHttpServerRequest&, QHttpServerResponder& responder) { QHttpServerResponse resp(QJsonObject{{QStringLiteral("error"), QStringLiteral("not_found")}}, diff --git a/src/api/ipc_dbus/CommandDispatcher.cpp b/src/api/ipc_dbus/CommandDispatcher.cpp index c8250ad..8a724b8 100644 --- a/src/api/ipc_dbus/CommandDispatcher.cpp +++ b/src/api/ipc_dbus/CommandDispatcher.cpp @@ -1,6 +1,8 @@ #include "api/ipc_dbus/CommandDispatcher.h" -#include "message_bus/pipeline/DynamicRuleRegistry.h" +#include "message_bus/pipeline/rules/DynamicRuleRegistry.h" +#include "message_bus/pipeline/discovery/DiscoveryPool.h" +#include "message_bus/pipeline/mapping/MappingRegistry.h" namespace softbus::api::ipc { @@ -22,6 +24,8 @@ softbus::message_bus::pipeline::RuleScope parseScope(const QJsonObject& params) DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObject& params) const { auto& reg = softbus::message_bus::pipeline::DynamicRuleRegistry::instance(); + auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); + auto& discovery = softbus::message_bus::pipeline::DiscoveryPool::instance(); if (action == QStringLiteral("add_filter")) { softbus::message_bus::pipeline::FilterRule rule; rule.id = params.value(QStringLiteral("id")).toString(); @@ -57,7 +61,45 @@ DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObj QJsonObject{{QStringLiteral("id"), id}}}; } if (action == QStringLiteral("list_rules") || action == QStringLiteral("get_status")) { - return {0, QStringLiteral("ok"), reg.status()}; + QJsonObject status = reg.status(); + status.insert(QStringLiteral("discovery"), discovery.status()); + status.insert(QStringLiteral("mapping"), mappingReg.status()); + return {0, QStringLiteral("ok"), status}; + } + if (action == QStringLiteral("add_mapping_rule")) { + softbus::message_bus::pipeline::MappingRule rule; + const QString keyText = params.value(QStringLiteral("compositeKey")).toString(); + rule.compositeKey = keyText.toULongLong(); + rule.domPath = params.value(QStringLiteral("domPath")).toString(); + rule.scale = params.value(QStringLiteral("scale")).toDouble(1.0); + rule.offset = params.value(QStringLiteral("offset")).toDouble(0.0); + rule.unit = params.value(QStringLiteral("unit")).toString(); + if (!mappingReg.upsert(rule)) { + return {4004, QStringLiteral("invalid_mapping_rule"), {}}; + } + return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("compositeKey"), keyText}}}; + } + if (action == QStringLiteral("remove_mapping_rule")) { + const QString keyText = params.value(QStringLiteral("compositeKey")).toString(); + const bool ok = mappingReg.remove(keyText.toULongLong()); + return {ok ? 0 : 4043, ok ? QStringLiteral("ok") : QStringLiteral("mapping_rule_not_found"), + QJsonObject{{QStringLiteral("compositeKey"), keyText}}}; + } + if (action == QStringLiteral("list_mapping_rules")) { + return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("items"), mappingReg.list()}}}; + } + if (action == QStringLiteral("discovery_start")) { + const auto ttlMs = params.value(QStringLiteral("ttlMs")).toInteger(0); + discovery.startSniffing(ttlMs); + return {0, QStringLiteral("ok"), discovery.status()}; + } + if (action == QStringLiteral("discovery_stop")) { + discovery.stopSniffing(); + return {0, QStringLiteral("ok"), discovery.status()}; + } + if (action == QStringLiteral("discovery_clear")) { + discovery.clear(); + return {0, QStringLiteral("ok"), discovery.status()}; } return {4040, QStringLiteral("unknown_action"), {}}; } diff --git a/src/core/plugin_system/IMapperDependencies.h b/src/core/plugin_system/IMapperDependencies.h new file mode 100644 index 0000000..0fc16b0 --- /dev/null +++ b/src/core/plugin_system/IMapperDependencies.h @@ -0,0 +1,54 @@ +#pragma once + +#include +#include + +#include +#include + +#include "core/models/RawBusMessage.h" + +namespace softbus::core::plugin_system +{ + +struct MappingLookupResult +{ + QString domPath; + double scale{1.0}; + double offset{0.0}; + QString unit; +}; + +struct DiscoveryRecord +{ + 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 lastSeenMs{0}; +}; + +class IMappingLookup +{ +public: + virtual ~IMappingLookup() = default; + virtual std::optional find(std::uint64_t compositeKey) const = 0; +}; + +class IDiscoverySink +{ +public: + virtual ~IDiscoverySink() = default; + virtual void reportUnmapped(const DiscoveryRecord& record) = 0; +}; + +struct MapperSessionContext +{ + IMappingLookup* mappingLookup{nullptr}; + IDiscoverySink* discoverySink{nullptr}; + QJsonObject params; +}; + +} // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/IProtocolMapperPlugin.h b/src/core/plugin_system/IProtocolMapperPlugin.h index 9902a01..919cf1c 100644 --- a/src/core/plugin_system/IProtocolMapperPlugin.h +++ b/src/core/plugin_system/IProtocolMapperPlugin.h @@ -7,6 +7,7 @@ #include "core/models/DOMMessage.h" #include "core/models/RawBusMessage.h" +#include "core/plugin_system/IMapperDependencies.h" #include "message_bus/pipeline/protocol/ProtocolEnvelope.h" namespace softbus::core::plugin_system @@ -30,6 +31,11 @@ public: virtual QString pluginId() const = 0; virtual softbus::core::models::ProtocolType protocolType() const = 0; virtual std::unique_ptr createMapperSession() = 0; + virtual std::unique_ptr createMapperSession(const MapperSessionContext& ctx) + { + (void)ctx; + return createMapperSession(); + } }; } // namespace softbus::core::plugin_system diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index 0bb3820..c8ffbfb 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -6,12 +6,15 @@ #include #include #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 chunkQ; @@ -32,6 +35,41 @@ struct EndpointFramingState namespace { +class MappingLookupAdapter final : public softbus::core::plugin_system::IMappingLookup +{ +public: + std::optional 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 st) { while (st->running.load(std::memory_order_acquire)) { @@ -515,7 +553,16 @@ std::vector PipelineEngine::stage2Parser(cons std::lock_guard 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(); + } + if (!m_discoverySink) { + m_discoverySink = std::make_unique(); + } + 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(std::move(s)) : nullptr); diff --git a/src/message_bus/pipeline/PipelineEngine.h b/src/message_bus/pipeline/PipelineEngine.h index 3d8b983..7cad5d2 100644 --- a/src/message_bus/pipeline/PipelineEngine.h +++ b/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> m_parserSessions; std::mutex m_mapperSessionMutex; QHash> m_mapperSessions; + std::unique_ptr m_mappingLookup; + std::unique_ptr m_discoverySink; std::mutex m_seqMutex; QHash m_nextFrameSeqByEndpoint; diff --git a/src/message_bus/pipeline/discovery/DiscoveryPool.cpp b/src/message_bus/pipeline/discovery/DiscoveryPool.cpp new file mode 100644 index 0000000..637a3cf --- /dev/null +++ b/src/message_bus/pipeline/discovery/DiscoveryPool.cpp @@ -0,0 +1,214 @@ +#include "message_bus/pipeline/discovery/DiscoveryPool.h" + +#include + +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(m_slots.size())); + o.insert(QStringLiteral("maxEntries"), static_cast(m_cfg.maxEntries)); + o.insert(QStringLiteral("minSampleIntervalMs"), static_cast(m_cfg.minSampleIntervalMs)); + o.insert(QStringLiteral("sniffStartedAtMs"), static_cast(m_sniffStartedAtMs)); + o.insert(QStringLiteral("sniffExpiresAtMs"), static_cast(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& mappedKeys) const +{ + QMutexLocker lk(&m_mtx); + std::vector 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(s.endpointHash)); + o.insert(QStringLiteral("protocol"), protocolToString(s.protocol)); + o.insert(QStringLiteral("slaveId"), static_cast(s.slaveId)); + o.insert(QStringLiteral("registerAddress"), static_cast(s.registerAddress)); + o.insert(QStringLiteral("rawValue"), s.rawValue); + o.insert(QStringLiteral("firstSeenMs"), static_cast(s.firstSeenMs)); + o.insert(QStringLiteral("lastSeenMs"), static_cast(s.lastSeenMs)); + o.insert(QStringLiteral("hitCount"), static_cast(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 diff --git a/src/message_bus/pipeline/discovery/DiscoveryPool.h b/src/message_bus/pipeline/discovery/DiscoveryPool.h new file mode 100644 index 0000000..8dfa6ca --- /dev/null +++ b/src/message_bus/pipeline/discovery/DiscoveryPool.h @@ -0,0 +1,81 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#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& 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 m_slots; + Config m_cfg; + std::atomic m_sniffingEnabled{false}; + std::int64_t m_sniffStartedAtMs{0}; + std::int64_t m_sniffExpiresAtMs{0}; +}; + +} // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/mapping/MappingRegistry.cpp b/src/message_bus/pipeline/mapping/MappingRegistry.cpp new file mode 100644 index 0000000..15dfc1a --- /dev/null +++ b/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 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 MappingRegistry::snapshotMappedKeys() const +{ + QReadLocker lk(&m_lock); + std::unordered_map 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(m_rules.size())}}; +} + +} // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/mapping/MappingRegistry.h b/src/message_bus/pipeline/mapping/MappingRegistry.h new file mode 100644 index 0000000..bcc41ab --- /dev/null +++ b/src/message_bus/pipeline/mapping/MappingRegistry.h @@ -0,0 +1,45 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include +#include + +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 find(std::uint64_t compositeKey) const; + + std::unordered_map snapshotMappedKeys() const; + QJsonArray list() const; + QJsonObject status() const; + +private: + MappingRegistry() = default; + +private: + mutable QReadWriteLock m_lock; + std::unordered_map m_rules; +}; + +} // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/DynamicRuleRegistry.cpp b/src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp similarity index 95% rename from src/message_bus/pipeline/DynamicRuleRegistry.cpp rename to src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp index f3f059b..bf7d5f4 100644 --- a/src/message_bus/pipeline/DynamicRuleRegistry.cpp +++ b/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 #include @@ -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 - diff --git a/src/message_bus/pipeline/DynamicRuleRegistry.h b/src/message_bus/pipeline/rules/DynamicRuleRegistry.h similarity index 99% rename from src/message_bus/pipeline/DynamicRuleRegistry.h rename to src/message_bus/pipeline/rules/DynamicRuleRegistry.h index bda1298..b74c3dc 100644 --- a/src/message_bus/pipeline/DynamicRuleRegistry.h +++ b/src/message_bus/pipeline/rules/DynamicRuleRegistry.h @@ -60,4 +60,3 @@ private: }; } // namespace softbus::message_bus::pipeline -