diff --git a/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx b/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx index 6ac8de8..9c8b885 100644 Binary files a/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx and b/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx differ diff --git a/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx b/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx index 3d4ccdd..51b6cfb 100644 Binary files a/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx and b/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx differ diff --git a/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx b/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx index 92dace8..f579254 100644 Binary files a/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx and b/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx differ diff --git a/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx b/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx index f58f38e..2f3059f 100644 Binary files a/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx and b/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx differ diff --git a/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx b/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx index e8a76d3..573bfe2 100644 Binary files a/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx and b/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx differ diff --git a/.cache/clangd/index/DeviceTypes.h.265DE4F5160B4435.idx b/.cache/clangd/index/DeviceTypes.h.265DE4F5160B4435.idx index 67fdc04..33edfbf 100644 Binary files a/.cache/clangd/index/DeviceTypes.h.265DE4F5160B4435.idx and b/.cache/clangd/index/DeviceTypes.h.265DE4F5160B4435.idx differ diff --git a/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx b/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx index db8776b..5b4efef 100644 Binary files a/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx and b/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx differ diff --git a/CMakeLists.txt b/CMakeLists.txt index 9b3d768..5022455 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -160,6 +160,8 @@ qt_add_executable(softbus_daemon src/message_bus/pipeline/mapping/MappingRegistry.cpp src/message_bus/pipeline/PipelineEngine.h src/message_bus/pipeline/PipelineEngine.cpp + src/message_bus/pipeline/PipelineEngineApi.h + src/message_bus/pipeline/PipelineEngineApi.cpp src/message_bus/pipeline/protocol/ProtocolEnvelope.h plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h diff --git a/bin/softbus_daemon b/bin/softbus_daemon index 666a12d..2ceb699 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp index 23ca7fb..d0dfa0d 100644 --- a/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp +++ b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp @@ -24,7 +24,7 @@ public: { m_buf.resize(m_cfg.maxAdu); } - + // 功能 ： 处理收到的数据块 void feedChunk(softbus::core::models::RawBusMessage& in, std::vector& out) override { @@ -47,6 +47,7 @@ public: m_lastActivity = m_lastRx; if (m_len + n > m_cfg.maxAdu) { + // 功能 ： 如果数据块超过最大长度，则丢弃 LOG_WARNING() << "ModbusRtuFramerPlugin: overflow endpointHash=0x" << QString::number(m_endpointHash, 16); m_len = 0; diff --git a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp index c8485ad..38040b0 100644 --- a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp +++ b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp @@ -64,6 +64,19 @@ public: rec.lastSeenMs = rawContext.timestampMs; m_context.discoverySink->reportUnmapped(rec); } + // 无 metadata 映射时仍产出 DOM，便于 DynamicRuleRegistry 的 transform 等下游逻辑生效 + // （compositeKey 仍按与映射表相同的算法计算，便于后续在 UI 里绑定映射） + softbus::core::models::DOMMessage pass; + pass.messageId = QStringLiteral("modbus.unmapped:%1").arg(compKey); + pass.timestamp = rawContext.timestampMs; + pass.value = rawValue; + pass.quality = softbus::core::models::DataQuality::GOOD; + pass.attributes.insert(QStringLiteral("sourceEndpoint"), QString::number(rawContext.endpointHash)); + pass.attributes.insert(QStringLiteral("compositeKey"), QString::number(compKey)); + pass.attributes.insert(QStringLiteral("slaveId"), static_cast(resp->unitId)); + pass.attributes.insert(QStringLiteral("registerAddress"), static_cast(addr)); + pass.attributes.insert(QStringLiteral("unmapped"), true); + out.push_back(std::move(pass)); continue; } softbus::core::models::DOMMessage msg; diff --git a/src/api/API_INTERFACE_TABLE.md b/src/api/API_INTERFACE_TABLE.md index a6de5e9..945a84b 100644 --- a/src/api/API_INTERFACE_TABLE.md +++ b/src/api/API_INTERFACE_TABLE.md @@ -78,7 +78,57 @@ | `GET` | `/v1/status` | 透传 `get_status` 的状态响应 | | `POST` | `/v1/command` | 统一命令入口（请求体为 executeCommand JSON） | -### 5.2 Discovery 路由 +### 5.2 设备树 / 设备路由 + +| 方法 | 路径 | 请求参数 | 说明 | +|---|---|---|---| +| `GET` | `/api/v1/device-tree` | 无 | 返回当前设备树快照 | +| `GET` | `/api/v1/devices` | 无 | 返回当前运行态设备列表 | +| `GET` | `/api/v1/devices/{id}` | path: `id(int)` | 返回指定运行态设备详情 | +| `PATCH` | `/api/v1/device-tree/nodes/{id}` | body: 任意可变节点字段子集 | 部分更新设备树节点；当前至少保护 `id`、`params.stableKey`、`params.online`、`params.lastSeenTs` | + +### 5.3 规则管理路由 + +| 方法 | 路径 | 请求参数 | 说明 | +|---|---|---|---| +| `GET` | `/api/v1/rules/status` | 无 | 返回动态规则计数状态 | +| `POST` | `/api/v1/rules/filter` | body: `id`, `endpointHash?`, `deviceId?`, `stableKey?`, `condition?` | 新增过滤规则 | +| `DELETE` | `/api/v1/rules/filter/{id}` | path: `id` | 删除过滤规则 | +| `POST` | `/api/v1/rules/transform` | body: `id`, `endpointHash?`, `deviceId?`, `stableKey?`, `scale?`, `offset?`, `unit?` | 新增转换规则 | +| `DELETE` | `/api/v1/rules/transform/{id}` | path: `id` | 删除转换规则 | + +- 说明： + - `transform` 规则已经在 `PipelineEngine::applyTransformAndEnrich(...)` 中生效。 + - `filter` 路由当前主要提供“注册表管理”能力；动态 filter 规则尚未真正接入 `PipelineEngine::stage3Filter(...)` 的执行链路。 + +### 5.4 Pipeline 路由 + +| 方法 | 路径 | 请求参数 | 说明 | +|---|---|---|---| +| `GET` | `/api/v1/pipeline/status` | 无 | 返回 pipeline 运行状态、session 数量、队列容量、插件摘要与 counters | +| `GET` | `/api/v1/pipeline/counters` | 无 | 返回 `in/out/drop/error/mergeDrop` 计数器快照 | +| `GET` | `/api/v1/pipeline/config` | 无 | 返回当前运行时配置快照 | + +- 说明：`/v1/status` 仍然是 control-plane 摘要（规则 + discovery + mapping），不等同于 `/api/v1/pipeline/status`。 + +### 5.5 插件 / 协议能力路由 + +| 方法 | 路径 | 请求参数 | 说明 | +|---|---|---|---| +| `GET` | `/api/v1/plugins` | 无 | 返回 parser / framer / mapper 三类逻辑插件的统一列表 | +| `GET` | `/api/v1/plugins/protocols` | 无 | 返回按协议聚合后的能力快照（parser/framer/mapper pluginId） | +| `GET` | `/api/v1/plugins/mappers` | 无 | 返回 mapper 插件列表 | + +### 5.6 Metadata / Profile 路由 + +| 方法 | 路径 | 请求参数 | 说明 | +|---|---|---|---| +| `GET` | `/api/v1/metadata` | 无 | 返回 metadata 字典全量快照 | +| `GET` | `/api/v1/metadata/{metadataId}` | path: `metadataId` | 返回指定 metadata 详情 | +| `GET` | `/api/v1/profiles` | 无 | 返回全部设备 profile 快照 | +| `GET` | `/api/v1/profiles/{deviceId}` | path: `deviceId` | 返回指定 `deviceId` 关联的全部 profile | + +### 5.7 Discovery 路由 | 方法 | 路径 | 请求参数 | 说明 | |---|---|---|---| @@ -88,7 +138,7 @@ | `POST` | `/api/v1/discovery/clear` | 无 | 清空发现池样本 | | `GET` | `/api/v1/discovery/samples` | query: `offset?`, `limit?`, `endpointHash?`, `protocol?` | 拉取样本（按更新时间倒序，且过滤已映射 key） | -### 5.3 Mapping 路由 +### 5.8 Mapping 路由 | 方法 | 路径 | 请求参数 | 说明 | |---|---|---|---| diff --git a/src/api/http/RestApiRoutes.cpp b/src/api/http/RestApiRoutes.cpp index 6474509..8f0e248 100644 --- a/src/api/http/RestApiRoutes.cpp +++ b/src/api/http/RestApiRoutes.cpp @@ -1,7 +1,12 @@ #include "api/http/RestApiRoutes.h" #include +#include +#include +#include +#include #include +#include #include #include #include @@ -9,13 +14,20 @@ #include #include #include +#include #include #include #include "api/http/RestApiCommandHttp.h" #include "api/ipc_dbus/CommandDispatcher.h" +#include "core/metadata/MetadataRegistry.h" +#include "core/metadata/ProfileRegistry.h" +#include "core/plugin_system/PluginManager.h" +#include "device_bus/DeviceBus.h" #include "message_bus/pipeline/discovery/DiscoveryPool.h" #include "message_bus/pipeline/mapping/MappingRegistry.h" +#include "message_bus/pipeline/PipelineEngine.h" +#include "message_bus/pipeline/rules/DynamicRuleRegistry.h" namespace softbus::api::http { @@ -23,6 +35,97 @@ namespace softbus::api::http namespace { +QHttpServerResponse withCorsHeaders(QHttpServerResponse response) +{ + QHttpHeaders headers = response.headers(); + headers.append(QHttpHeaders::WellKnownHeader::AccessControlAllowOrigin, QByteArrayView("*")); + headers.append(QHttpHeaders::WellKnownHeader::AccessControlAllowMethods, + QByteArrayView("GET, POST, PATCH, DELETE, OPTIONS")); + headers.append(QHttpHeaders::WellKnownHeader::AccessControlAllowHeaders, + QByteArrayView("Content-Type, Authorization, X-Requested-With")); + headers.append(QHttpHeaders::WellKnownHeader::AccessControlMaxAge, QByteArrayView("86400")); + response.setHeaders(std::move(headers)); + return response; +} + +QHttpServerResponse corsPreflightResponse() +{ + return withCorsHeaders(QHttpServerResponse(QHttpServerResponse::StatusCode::NoContent)); +} + +void registerCorsPreflightRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/health"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/v1/status"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/v1/command"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/discovery/start"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/discovery/stop"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/discovery/status"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/discovery/clear"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/discovery/samples"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/mapping/bind"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/mapping/unbind"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/mapping/list"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/device-tree"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/devices"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/devices/"), QHttpServerRequest::Method::Options, routeContext, + [](const QString&) { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/device-tree/nodes/"), QHttpServerRequest::Method::Options, routeContext, + [](const QString&) { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/rules/status"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/rules/filter"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/rules/filter/"), QHttpServerRequest::Method::Options, routeContext, + [](const QString&) { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/rules/transform"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/rules/transform/"), QHttpServerRequest::Method::Options, routeContext, + [](const QString&) { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/pipeline/status"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/pipeline/counters"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/pipeline/counters/clear"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/pipeline/config"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/plugins"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/plugins/protocols"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/plugins/mappers"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/metadata"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/metadata/"), QHttpServerRequest::Method::Options, routeContext, + [](const QString&) { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/profiles"), QHttpServerRequest::Method::Options, routeContext, + []() { return corsPreflightResponse(); }); + server->route(QStringLiteral("/api/v1/profiles/"), QHttpServerRequest::Method::Options, routeContext, + [](const QString&) { return corsPreflightResponse(); }); +} + +constexpr auto kProtectedNodeKeys = std::array{ + "id", + "stableKey", + "online", + "lastSeenTs", +}; + softbus::core::models::ProtocolType protocolFromString(const QString& s) { const QString x = s.trimmed().toUpper(); @@ -47,16 +150,190 @@ softbus::core::models::ProtocolType protocolFromString(const QString& s) return softbus::core::models::ProtocolType::UNKNOWN; } -QJsonObject parseJsonBody(const QHttpServerRequest& request) +QJsonObject parseJsonBody(const QHttpServerRequest& request, bool* ok = nullptr) { QJsonParseError err; const auto doc = QJsonDocument::fromJson(request.body(), &err); - if (err.error != QJsonParseError::NoError || !doc.isObject()) { + const bool parsed = err.error == QJsonParseError::NoError && doc.isObject(); + if (ok) { + *ok = parsed; + } + if (!parsed) { return {}; } return doc.object(); } +QHttpServerResponse jsonResponse(const QJsonObject& obj, + QHttpServerResponse::StatusCode code = QHttpServerResponse::StatusCode::Ok) +{ + return withCorsHeaders(QHttpServerResponse(obj, code)); +} + +QHttpServerResponse okDataResponse(const QJsonValue& data, + QHttpServerResponse::StatusCode code = QHttpServerResponse::StatusCode::Ok) +{ + return jsonResponse(QJsonObject{{QStringLiteral("ok"), true}, {QStringLiteral("data"), data}}, code); +} + +QHttpServerResponse errorResponse(const QString& error, + QHttpServerResponse::StatusCode code = QHttpServerResponse::StatusCode::BadRequest) +{ + return jsonResponse(QJsonObject{{QStringLiteral("ok"), false}, {QStringLiteral("error"), error}}, code); +} + +QJsonObject deviceTreeNodeToJson(const DeviceTreeNode& node) +{ + QJsonObject obj; + obj.insert(QStringLiteral("id"), node.id); + obj.insert(QStringLiteral("name"), node.name); + obj.insert(QStringLiteral("type"), node.type); + obj.insert(QStringLiteral("endpoint"), node.endpoint); + obj.insert(QStringLiteral("driver"), node.driver); + obj.insert(QStringLiteral("capabilitiesRef"), node.capabilitiesRef); + obj.insert(QStringLiteral("params"), node.params); + obj.insert(QStringLiteral("enabled"), node.enabled); + QJsonArray children; + for (const auto& child : node.childrenIds) { + children.push_back(child); + } + obj.insert(QStringLiteral("children"), children); + return obj; +} + +QJsonArray devicesToJson(const std::vector& devices) +{ + QJsonArray out; + for (const auto& device : devices) { + out.push_back(device.toJsonObject()); + } + return out; +} + +QString resolveConfigBaseDir() +{ + QString configBase = qEnvironmentVariable("SOFTBUS_CONFIG_DIR"); + if (!configBase.isEmpty()) { + return configBase.endsWith(QChar('/')) ? configBase : configBase + QChar('/'); + } + return QDir::homePath() + QStringLiteral("/softbus/config/"); +} + +QString resolveConfigFile(const QString& fileName, const QString& fallbackRelative) +{ + const QString homePath = resolveConfigBaseDir() + fileName; + return QFileInfo::exists(homePath) ? homePath : fallbackRelative; +} + +bool ensureMetadataLoaded(softbus::core::metadata::MetadataRegistry& registry, QString* error) +{ + static std::mutex mtx; + static bool attempted = false; + static bool loaded = false; + std::lock_guard lk(mtx); + if (!attempted) { + attempted = true; + loaded = registry.loadFromFile( + resolveConfigFile(QStringLiteral("metadata_dictionary.json"), + QStringLiteral("config/metadata_dictionary.json"))); + } + if (!loaded && error) { + *error = registry.lastError(); + } + return loaded; +} + +softbus::core::metadata::MetadataRegistry& metadataRegistry() +{ + static softbus::core::metadata::MetadataRegistry registry; + return registry; +} + +bool ensureProfileLoaded(softbus::core::metadata::ProfileRegistry& registry, QString* error) +{ + static std::mutex mtx; + static bool attempted = false; + static bool loaded = false; + std::lock_guard lk(mtx); + if (!attempted) { + attempted = true; + loaded = registry.loadFromFile( + resolveConfigFile(QStringLiteral("device_profiles.json"), + QStringLiteral("config/device_profiles.json"))); + } + if (!loaded && error) { + *error = registry.lastError(); + } + return loaded; +} + +softbus::core::metadata::ProfileRegistry& profileRegistry() +{ + static softbus::core::metadata::ProfileRegistry registry; + return registry; +} + +bool isProtectedNodeKey(const QString& key) +{ + for (const auto* value : kProtectedNodeKeys) { + if (key == QLatin1String(value)) { + return true; + } + } + return false; +} + +bool applyNodePatch(DeviceTreeNode& node, const QJsonObject& patch) +{ + bool changed = false; + + if (patch.contains(QStringLiteral("name")) && patch.value(QStringLiteral("name")).isString()) { + node.name = patch.value(QStringLiteral("name")).toString(); + changed = true; + } + if (patch.contains(QStringLiteral("type")) && patch.value(QStringLiteral("type")).isString()) { + node.type = patch.value(QStringLiteral("type")).toString(); + changed = true; + } + if (patch.contains(QStringLiteral("endpoint")) && patch.value(QStringLiteral("endpoint")).isString()) { + node.endpoint = patch.value(QStringLiteral("endpoint")).toString(); + changed = true; + } + if (patch.contains(QStringLiteral("driver")) && patch.value(QStringLiteral("driver")).isString()) { + node.driver = patch.value(QStringLiteral("driver")).toString(); + changed = true; + } + if (patch.contains(QStringLiteral("capabilitiesRef")) && patch.value(QStringLiteral("capabilitiesRef")).isString()) { + node.capabilitiesRef = patch.value(QStringLiteral("capabilitiesRef")).toString(); + changed = true; + } + if (patch.contains(QStringLiteral("enabled")) && patch.value(QStringLiteral("enabled")).isBool()) { + node.enabled = patch.value(QStringLiteral("enabled")).toBool(); + changed = true; + } + if (patch.contains(QStringLiteral("children")) && patch.value(QStringLiteral("children")).isArray()) { + QVector children; + for (const auto& child : patch.value(QStringLiteral("children")).toArray()) { + if (child.isString()) { + children.push_back(child.toString()); + } + } + node.childrenIds = children; + changed = true; + } + if (patch.contains(QStringLiteral("params")) && patch.value(QStringLiteral("params")).isObject()) { + const auto paramsPatch = patch.value(QStringLiteral("params")).toObject(); + for (auto it = paramsPatch.begin(); it != paramsPatch.end(); ++it) { + if (isProtectedNodeKey(it.key())) { + continue; + } + node.params.insert(it.key(), it.value()); + changed = true; + } + } + return changed; +} + void registerDiscoveryRoutes(QHttpServer* server, QObject* routeContext) { server->route(QStringLiteral("/api/v1/discovery/start"), QHttpServerRequest::Method::Post, routeContext, @@ -65,25 +342,25 @@ void registerDiscoveryRoutes(QHttpServer* server, QObject* routeContext) 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); + return withCorsHeaders(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); + return withCorsHeaders(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); + return withCorsHeaders(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); + return withCorsHeaders(QHttpServerResponse(pool.status(), QHttpServerResponse::StatusCode::Ok)); }); server->route(QStringLiteral("/api/v1/discovery/samples"), QHttpServerRequest::Method::Get, routeContext, @@ -108,7 +385,7 @@ void registerDiscoveryRoutes(QHttpServer* server, QObject* routeContext) 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); + return withCorsHeaders(QHttpServerResponse(o, QHttpServerResponse::StatusCode::Ok)); }); } @@ -126,14 +403,14 @@ void registerMappingRoutes(QHttpServer* server, QObject* routeContext) auto& mappingReg = softbus::message_bus::pipeline::MappingRegistry::instance(); if (!mappingReg.upsert(rule)) { - return QHttpServerResponse( + return withCorsHeaders(QHttpServerResponse( QJsonObject{{QStringLiteral("error"), QStringLiteral("invalid_mapping_rule")}}, - QHttpServerResponse::StatusCode::BadRequest); + QHttpServerResponse::StatusCode::BadRequest)); } - return QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), true}, + return withCorsHeaders(QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), true}, {QStringLiteral("compositeKey"), QString::number(rule.compositeKey)}}, - QHttpServerResponse::StatusCode::Ok); + QHttpServerResponse::StatusCode::Ok)); }); server->route(QStringLiteral("/api/v1/mapping/unbind"), QHttpServerRequest::Method::Post, routeContext, @@ -143,20 +420,264 @@ void registerMappingRoutes(QHttpServer* server, QObject* routeContext) 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}, + return withCorsHeaders(QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), ok}, {QStringLiteral("compositeKey"), QString::number(compositeKey)}}, ok ? QHttpServerResponse::StatusCode::Ok - : QHttpServerResponse::StatusCode::NotFound); + : 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); + return withCorsHeaders( + QHttpServerResponse(QJsonObject{{QStringLiteral("items"), mappingReg.list()}}, + QHttpServerResponse::StatusCode::Ok)); }); } +void registerDeviceRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/api/v1/device-tree"), QHttpServerRequest::Method::Get, routeContext, []() { + auto model = softbus::device_bus::DeviceBus::instance().deviceTreeModel(); + if (!model) { + return errorResponse(QStringLiteral("device_tree_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(model->toDeviceTreeJson()); + }); + + server->route(QStringLiteral("/api/v1/devices"), QHttpServerRequest::Method::Get, routeContext, []() { + return okDataResponse(devicesToJson(softbus::device_bus::DeviceBus::instance().listDevices())); + }); + + server->route(QStringLiteral("/api/v1/devices/"), QHttpServerRequest::Method::Get, routeContext, + [](int id) { + const auto device = softbus::device_bus::DeviceBus::instance().findDeviceById(id); + if (!device.has_value()) { + return errorResponse(QStringLiteral("device_not_found"), + QHttpServerResponse::StatusCode::NotFound); + } + return okDataResponse(device->toJsonObject()); + }); + + server->route(QStringLiteral("/api/v1/device-tree/nodes/"), QHttpServerRequest::Method::Patch, routeContext, + [](const QString& id, const QHttpServerRequest& request) { + bool bodyOk = false; + const auto patch = parseJsonBody(request, &bodyOk); + if (!bodyOk) { + return errorResponse(QStringLiteral("invalid_json"), + QHttpServerResponse::StatusCode::BadRequest); + } + + auto model = softbus::device_bus::DeviceBus::instance().deviceTreeModel(); + if (!model) { + return errorResponse(QStringLiteral("device_tree_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + const int index = model->findIndexById(id); + if (index < 0) { + return errorResponse(QStringLiteral("device_tree_node_not_found"), + QHttpServerResponse::StatusCode::NotFound); + } + + auto& node = model->nodesMutable()[index]; + const bool changed = applyNodePatch(node, patch); + if (!changed) { + return errorResponse(QStringLiteral("no_mutable_fields_provided"), + QHttpServerResponse::StatusCode::BadRequest); + } + + model->rebuildIndex(); + auto& bus = softbus::device_bus::DeviceBus::instance(); + bus.markDeviceTreeDirty(); + (void)bus.reconcileFromTreeModel(); + return okDataResponse(deviceTreeNodeToJson(model->nodes().at(index))); + }); +} + +void registerRuleRoutes(QHttpServer* server, + QObject* routeContext, + softbus::api::ipc::CommandDispatcher* dispatcher) +{ + server->route(QStringLiteral("/api/v1/rules/status"), QHttpServerRequest::Method::Get, routeContext, []() { + return okDataResponse(softbus::message_bus::pipeline::DynamicRuleRegistry::instance().status()); + }); + + server->route(QStringLiteral("/api/v1/rules/filter"), QHttpServerRequest::Method::Post, routeContext, + [dispatcher](const QHttpServerRequest& request) { + bool bodyOk = false; + const auto payload = parseJsonBody(request, &bodyOk); + if (!bodyOk) { + return errorResponse(QStringLiteral("invalid_json"), + QHttpServerResponse::StatusCode::BadRequest); + } + const auto result = dispatcher->dispatch(QStringLiteral("add_filter"), payload); + return withCorsHeaders(makeCommandHttpResponse(QString(), result)); + }); + + server->route(QStringLiteral("/api/v1/rules/filter/"), QHttpServerRequest::Method::Delete, routeContext, + [dispatcher](const QString& id) { + const auto result = dispatcher->dispatch(QStringLiteral("remove_filter"), + QJsonObject{{QStringLiteral("id"), id}}); + return withCorsHeaders(makeCommandHttpResponse(QString(), result)); + }); + + server->route(QStringLiteral("/api/v1/rules/transform"), QHttpServerRequest::Method::Post, routeContext, + [dispatcher](const QHttpServerRequest& request) { + bool bodyOk = false; + const auto payload = parseJsonBody(request, &bodyOk); + if (!bodyOk) { + return errorResponse(QStringLiteral("invalid_json"), + QHttpServerResponse::StatusCode::BadRequest); + } + const auto result = dispatcher->dispatch(QStringLiteral("add_transform_rule"), payload); + return withCorsHeaders(makeCommandHttpResponse(QString(), result)); + }); + + server->route(QStringLiteral("/api/v1/rules/transform/"), QHttpServerRequest::Method::Delete, routeContext, + [dispatcher](const QString& id) { + const auto result = dispatcher->dispatch(QStringLiteral("remove_transform_rule"), + QJsonObject{{QStringLiteral("id"), id}}); + return withCorsHeaders(makeCommandHttpResponse(QString(), result)); + }); +} + +void registerPipelineRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/api/v1/pipeline/status"), QHttpServerRequest::Method::Get, routeContext, []() { + auto* pipeline = softbus::device_bus::DeviceBus::instance().pipelineEngine(); + if (!pipeline) { + return errorResponse(QStringLiteral("pipeline_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(pipeline->status()); + }); + + server->route(QStringLiteral("/api/v1/pipeline/counters"), QHttpServerRequest::Method::Get, routeContext, []() { + auto* pipeline = softbus::device_bus::DeviceBus::instance().pipelineEngine(); + if (!pipeline) { + return errorResponse(QStringLiteral("pipeline_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(pipeline->countersJson()); + }); + + server->route(QStringLiteral("/api/v1/pipeline/counters/clear"), QHttpServerRequest::Method::Post, routeContext, []() { + auto* pipeline = softbus::device_bus::DeviceBus::instance().pipelineEngine(); + if (!pipeline) { + return errorResponse(QStringLiteral("pipeline_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + pipeline->resetCounters(); + return okDataResponse(pipeline->countersJson()); + }); + + server->route(QStringLiteral("/api/v1/pipeline/config"), QHttpServerRequest::Method::Get, routeContext, []() { + auto* pipeline = softbus::device_bus::DeviceBus::instance().pipelineEngine(); + if (!pipeline) { + return errorResponse(QStringLiteral("pipeline_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(pipeline->runtimeConfigJson()); + }); +} + +void registerPluginRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/api/v1/plugins"), QHttpServerRequest::Method::Get, routeContext, []() { + auto pluginManager = softbus::device_bus::DeviceBus::instance().pluginManager(); + if (!pluginManager) { + return errorResponse(QStringLiteral("plugin_manager_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + QJsonArray items; + for (const auto& entry : pluginManager->protocolPluginsSnapshot()) { + items.push_back(entry); + } + for (const auto& entry : pluginManager->protocolFramerPluginsSnapshot()) { + items.push_back(entry); + } + for (const auto& entry : pluginManager->protocolMapperPluginsSnapshot()) { + items.push_back(entry); + } + return okDataResponse(items); + }); + + server->route(QStringLiteral("/api/v1/plugins/protocols"), QHttpServerRequest::Method::Get, routeContext, []() { + auto pluginManager = softbus::device_bus::DeviceBus::instance().pluginManager(); + if (!pluginManager) { + return errorResponse(QStringLiteral("plugin_manager_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(pluginManager->protocolCapabilitiesSnapshot()); + }); + + server->route(QStringLiteral("/api/v1/plugins/mappers"), QHttpServerRequest::Method::Get, routeContext, []() { + auto pluginManager = softbus::device_bus::DeviceBus::instance().pluginManager(); + if (!pluginManager) { + return errorResponse(QStringLiteral("plugin_manager_unavailable"), + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(pluginManager->protocolMapperPluginsSnapshot()); + }); +} + +void registerMetadataRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/api/v1/metadata"), QHttpServerRequest::Method::Get, routeContext, []() { + auto& registry = metadataRegistry(); + QString error; + if (!ensureMetadataLoaded(registry, &error)) { + return errorResponse(error.isEmpty() ? QStringLiteral("metadata_unavailable") : error, + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(registry.list()); + }); + + server->route(QStringLiteral("/api/v1/metadata/"), QHttpServerRequest::Method::Get, routeContext, + [](const QString& metadataId) { + auto& registry = metadataRegistry(); + QString error; + if (!ensureMetadataLoaded(registry, &error)) { + return errorResponse(error.isEmpty() ? QStringLiteral("metadata_unavailable") : error, + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + const auto items = registry.list(); + for (const auto& item : items) { + const auto obj = item.toObject(); + if (obj.value(QStringLiteral("metadataId")).toString() == metadataId) { + return okDataResponse(obj); + } + } + return errorResponse(QStringLiteral("metadata_not_found"), + QHttpServerResponse::StatusCode::NotFound); + }); +} + +void registerProfileRoutes(QHttpServer* server, QObject* routeContext) +{ + server->route(QStringLiteral("/api/v1/profiles"), QHttpServerRequest::Method::Get, routeContext, []() { + auto& registry = profileRegistry(); + QString error; + if (!ensureProfileLoaded(registry, &error)) { + return errorResponse(error.isEmpty() ? QStringLiteral("profiles_unavailable") : error, + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(registry.list()); + }); + + server->route(QStringLiteral("/api/v1/profiles/"), QHttpServerRequest::Method::Get, routeContext, + [](const QString& deviceId) { + auto& registry = profileRegistry(); + QString error; + if (!ensureProfileLoaded(registry, &error)) { + return errorResponse(error.isEmpty() ? QStringLiteral("profiles_unavailable") : error, + QHttpServerResponse::StatusCode::ServiceUnavailable); + } + return okDataResponse(registry.listByDeviceId(deviceId)); + }); +} + } // namespace void registerSoftbusRestApiRoutes(QHttpServer* server, @@ -168,24 +689,33 @@ void registerSoftbusRestApiRoutes(QHttpServer* server, } server->route(QStringLiteral("/health"), QHttpServerRequest::Method::Get, routeContext, []() { - return QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), true}}, QHttpServerResponse::StatusCode::Ok); + return withCorsHeaders( + QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), true}}, QHttpServerResponse::StatusCode::Ok)); }); server->route(QStringLiteral("/v1/status"), QHttpServerRequest::Method::Get, routeContext, [dispatcher]() { - return makeCommandHttpResponse(QString(), dispatcher->dispatch(QStringLiteral("get_status"), {})); + return withCorsHeaders(makeCommandHttpResponse(QString(), dispatcher->dispatch(QStringLiteral("get_status"), {}))); }); server->route(QStringLiteral("/v1/command"), QHttpServerRequest::Method::Post, routeContext, [dispatcher](const QHttpServerRequest& request) { - return handleExecuteCommandHttpBody(request.body(), *dispatcher); + return withCorsHeaders(handleExecuteCommandHttpBody(request.body(), *dispatcher)); }); + + registerCorsPreflightRoutes(server, routeContext); registerDiscoveryRoutes(server, routeContext); registerMappingRoutes(server, routeContext); + registerDeviceRoutes(server, routeContext); + registerRuleRoutes(server, routeContext, dispatcher); + registerPipelineRoutes(server, routeContext); + registerPluginRoutes(server, routeContext); + registerMetadataRoutes(server, routeContext); + registerProfileRoutes(server, routeContext); server->setMissingHandler(routeContext, [](const QHttpServerRequest&, QHttpServerResponder& responder) { QHttpServerResponse resp(QJsonObject{{QStringLiteral("error"), QStringLiteral("not_found")}}, QHttpServerResponse::StatusCode::NotFound); - responder.sendResponse(std::move(resp)); + responder.sendResponse(withCorsHeaders(std::move(resp))); }); } diff --git a/src/api/ipc_dbus/CommandDispatcher.cpp b/src/api/ipc_dbus/CommandDispatcher.cpp index 8a724b8..820341f 100644 --- a/src/api/ipc_dbus/CommandDispatcher.cpp +++ b/src/api/ipc_dbus/CommandDispatcher.cpp @@ -1,8 +1,17 @@ #include "api/ipc_dbus/CommandDispatcher.h" +#include +#include +#include +#include +#include +#include + +#include "device_bus/DeviceBus.h" #include "message_bus/pipeline/rules/DynamicRuleRegistry.h" #include "message_bus/pipeline/discovery/DiscoveryPool.h" #include "message_bus/pipeline/mapping/MappingRegistry.h" +#include "utils/logs/logging.h" namespace softbus::api::ipc { @@ -10,6 +19,16 @@ namespace softbus::api::ipc namespace { +QString rulesConfigPath() +{ + const QString daemonConfigPath = softbus::device_bus::DeviceBus::instance().configPath(); + if (!daemonConfigPath.isEmpty()) { + const QFileInfo fi(daemonConfigPath); + return fi.dir().filePath(QStringLiteral("runtime_rules.json")); + } + return QDir::homePath() + QStringLiteral("/softbus/config/runtime_rules.json"); +} + softbus::message_bus::pipeline::RuleScope parseScope(const QJsonObject& params) { softbus::message_bus::pipeline::RuleScope scope; @@ -19,6 +38,89 @@ softbus::message_bus::pipeline::RuleScope parseScope(const QJsonObject& params) return scope; } +QJsonObject scopeToJson(const softbus::message_bus::pipeline::RuleScope& scope) +{ + QJsonObject out; + if (scope.endpointHash != 0) { + out.insert(QStringLiteral("endpointHash"), static_cast(scope.endpointHash)); + } + if (scope.deviceId >= 0) { + out.insert(QStringLiteral("deviceId"), scope.deviceId); + } + if (!scope.stableKey.isEmpty()) { + out.insert(QStringLiteral("stableKey"), scope.stableKey); + } + return out; +} + +QJsonObject filterRuleToJson(const softbus::message_bus::pipeline::FilterRule& rule) +{ + QJsonObject out; + out.insert(QStringLiteral("id"), rule.id); + out.insert(QStringLiteral("scope"), scopeToJson(rule.scope)); + out.insert(QStringLiteral("condition"), rule.condition); + return out; +} + +QJsonObject transformRuleToJson(const softbus::message_bus::pipeline::TransformRule& rule) +{ + QJsonObject out; + out.insert(QStringLiteral("id"), rule.id); + out.insert(QStringLiteral("scope"), scopeToJson(rule.scope)); + out.insert(QStringLiteral("scale"), rule.scale); + out.insert(QStringLiteral("offset"), rule.offset); + if (!rule.unit.isEmpty()) { + out.insert(QStringLiteral("unit"), rule.unit); + } + return out; +} + +bool persistRulesToDedicatedConfig(const softbus::message_bus::pipeline::DynamicRuleRegistry& reg) +{ + const QString configPath = rulesConfigPath(); + if (configPath.isEmpty()) { + LOG_ERROR() << "CommandDispatcher: rules config path is empty, cannot persist rules"; + return false; + } + + const QFileInfo fi(configPath); + QJsonArray filtersJson; + for (const auto& rule : reg.snapshotAllFilterRules()) { + filtersJson.push_back(filterRuleToJson(rule)); + } + QJsonArray transformsJson; + for (const auto& rule : reg.snapshotAllTransformRules()) { + transformsJson.push_back(transformRuleToJson(rule)); + } + + QJsonObject root; + root.insert(QStringLiteral("filter"), filtersJson); + root.insert(QStringLiteral("transform"), transformsJson); + + if (!QDir().mkpath(fi.dir().absolutePath())) { + LOG_ERROR() << "CommandDispatcher: failed to create rules config directory:" << fi.dir().absolutePath(); + return false; + } + + const QByteArray bytes = QJsonDocument(root).toJson(QJsonDocument::Indented); + QSaveFile out(configPath); + if (!out.open(QIODevice::WriteOnly | QIODevice::Text)) { + LOG_ERROR() << "CommandDispatcher: failed to open rules config for write:" << configPath; + return false; + } + if (out.write(bytes) != bytes.size()) { + LOG_ERROR() << "CommandDispatcher: failed to write rules config:" << configPath; + out.cancelWriting(); + return false; + } + if (!out.commit()) { + LOG_ERROR() << "CommandDispatcher: failed to commit rules config:" << configPath; + return false; + } + LOG_INFO() << "CommandDispatcher: rules persisted to" << configPath; + return true; +} + } // namespace DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObject& params) const @@ -34,11 +136,17 @@ DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObj if (!reg.addFilterRule(rule)) { return {4001, QStringLiteral("invalid_filter_rule"), {}}; } + if (!persistRulesToDedicatedConfig(reg)) { + return {5001, QStringLiteral("persist_rules_failed"), {}}; + } return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("id"), rule.id}}}; } if (action == QStringLiteral("remove_filter")) { const QString id = params.value(QStringLiteral("id")).toString(); const bool ok = reg.removeFilterRule(id); + if (ok && !persistRulesToDedicatedConfig(reg)) { + return {5001, QStringLiteral("persist_rules_failed"), {}}; + } return {ok ? 0 : 4041, ok ? QStringLiteral("ok") : QStringLiteral("filter_not_found"), QJsonObject{{QStringLiteral("id"), id}}}; } @@ -52,11 +160,17 @@ DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObj if (!reg.addTransformRule(rule)) { return {4002, QStringLiteral("invalid_transform_rule"), {}}; } + if (!persistRulesToDedicatedConfig(reg)) { + return {5001, QStringLiteral("persist_rules_failed"), {}}; + } return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("id"), rule.id}}}; } if (action == QStringLiteral("remove_transform_rule")) { const QString id = params.value(QStringLiteral("id")).toString(); const bool ok = reg.removeTransformRule(id); + if (ok && !persistRulesToDedicatedConfig(reg)) { + return {5001, QStringLiteral("persist_rules_failed"), {}}; + } return {ok ? 0 : 4042, ok ? QStringLiteral("ok") : QStringLiteral("transform_rule_not_found"), QJsonObject{{QStringLiteral("id"), id}}}; } diff --git a/src/core/metadata/MetadataRegistry.cpp b/src/core/metadata/MetadataRegistry.cpp index 9f4ec50..e342f9e 100644 --- a/src/core/metadata/MetadataRegistry.cpp +++ b/src/core/metadata/MetadataRegistry.cpp @@ -65,6 +65,65 @@ DataType parseDataType(const QString& s) return DataType::FLOAT64; } +QString domainToString(DataDomain domain) +{ + switch (domain) { + case DataDomain::PROCESS_PARAM: + return QStringLiteral("PROCESS_PARAM"); + case DataDomain::CONTROL_CMD: + return QStringLiteral("CONTROL_CMD"); + case DataDomain::MONITOR_DATA: + return QStringLiteral("MONITOR_DATA"); + case DataDomain::HEALTH_DIAG: + return QStringLiteral("HEALTH_DIAG"); + case DataDomain::CONFIG_PARAM: + return QStringLiteral("CONFIG_PARAM"); + case DataDomain::UNKNOWN: + default: + return QStringLiteral("UNKNOWN"); + } +} + +QString dataTypeToString(DataType type) +{ + switch (type) { + case DataType::BOOL: + return QStringLiteral("BOOL"); + case DataType::INT32: + return QStringLiteral("INT32"); + case DataType::UINT32: + return QStringLiteral("UINT32"); + case DataType::INT64: + return QStringLiteral("INT64"); + case DataType::FLOAT32: + return QStringLiteral("FLOAT32"); + case DataType::FLOAT64: + return QStringLiteral("FLOAT64"); + case DataType::STRING: + return QStringLiteral("STRING"); + case DataType::BYTE_ARRAY: + return QStringLiteral("BYTE_ARRAY"); + default: + return QStringLiteral("FLOAT64"); + } +} + +QJsonObject toJsonObject(const softbus::core::models::MetadataDef& def) +{ + return QJsonObject{ + {QStringLiteral("metadataId"), def.metadataId}, + {QStringLiteral("name"), def.name}, + {QStringLiteral("description"), def.description}, + {QStringLiteral("domain"), domainToString(def.domain)}, + {QStringLiteral("semanticType"), dataTypeToString(def.semanticType)}, + {QStringLiteral("unit"), def.unit}, + {QStringLiteral("minValue"), def.minValue}, + {QStringLiteral("maxValue"), def.maxValue}, + {QStringLiteral("deadband"), def.deadband}, + {QStringLiteral("isWriteable"), def.isWriteable}, + }; +} + } // namespace bool MetadataRegistry::loadFromFile(const QString& filePath) @@ -114,6 +173,15 @@ const softbus::core::models::MetadataDef* MetadataRegistry::findById(const QStri return &it.value(); } +QJsonArray MetadataRegistry::list() const +{ + QJsonArray out; + for (auto it = m_defsById.cbegin(); it != m_defsById.cend(); ++it) { + out.push_back(toJsonObject(it.value())); + } + return out; +} + std::optional MetadataRegistry::parseOne(const QJsonObject& o) { softbus::core::models::MetadataDef def; diff --git a/src/core/metadata/MetadataRegistry.h b/src/core/metadata/MetadataRegistry.h index 9a1a4d6..4f8516e 100644 --- a/src/core/metadata/MetadataRegistry.h +++ b/src/core/metadata/MetadataRegistry.h @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -17,6 +18,7 @@ public: bool loadFromFile(const QString& filePath); const softbus::core::models::MetadataDef* findById(const QString& metadataId) const; + QJsonArray list() const; bool empty() const { return m_defsById.isEmpty(); } const QString& lastError() const { return m_lastError; } diff --git a/src/core/metadata/ProfileRegistry.cpp b/src/core/metadata/ProfileRegistry.cpp index f448078..b7cb506 100644 --- a/src/core/metadata/ProfileRegistry.cpp +++ b/src/core/metadata/ProfileRegistry.cpp @@ -16,6 +16,75 @@ QString normalizedProtocol(const QString& protocol) return protocol.trimmed().toLower(); } +QString dataTypeToString(softbus::core::models::DataType type) +{ + using softbus::core::models::DataType; + switch (type) { + case DataType::BOOL: + return QStringLiteral("BOOL"); + case DataType::INT32: + return QStringLiteral("INT32"); + case DataType::UINT32: + return QStringLiteral("UINT32"); + case DataType::INT64: + return QStringLiteral("INT64"); + case DataType::FLOAT32: + return QStringLiteral("FLOAT32"); + case DataType::FLOAT64: + return QStringLiteral("FLOAT64"); + case DataType::STRING: + return QStringLiteral("STRING"); + case DataType::BYTE_ARRAY: + return QStringLiteral("BYTE_ARRAY"); + default: + return QStringLiteral("INT32"); + } +} + +QString endiannessToString(softbus::core::models::Endianness endianness) +{ + using softbus::core::models::Endianness; + switch (endianness) { + case Endianness::LittleEndian: + return QStringLiteral("LITTLE_ENDIAN"); + case Endianness::BigEndianByteSwap: + return QStringLiteral("BIG_ENDIAN_BYTE_SWAP"); + case Endianness::BigEndian: + default: + return QStringLiteral("BIG_ENDIAN"); + } +} + +QJsonObject mappingToJsonObject(const softbus::core::models::DataPointMapping& mapping) +{ + return QJsonObject{ + {QStringLiteral("pointId"), mapping.pointId}, + {QStringLiteral("metadataId"), mapping.metadataId}, + {QStringLiteral("registerAddress"), static_cast(mapping.registerAddress)}, + {QStringLiteral("byteOffset"), static_cast(mapping.byteOffset)}, + {QStringLiteral("byteLength"), static_cast(mapping.byteLength)}, + {QStringLiteral("bitOffset"), static_cast(mapping.bitOffset)}, + {QStringLiteral("endianness"), endiannessToString(mapping.endianness)}, + {QStringLiteral("rawType"), dataTypeToString(mapping.rawType)}, + {QStringLiteral("scaleFactor"), mapping.scaleFactor}, + {QStringLiteral("offsetFactor"), mapping.offsetFactor}, + {QStringLiteral("expression"), mapping.expression}, + }; +} + +QJsonObject profileToJsonObject(const softbus::core::models::DeviceDataProfile& profile) +{ + QJsonArray mappings; + for (const auto& mapping : profile.mappings) { + mappings.push_back(mappingToJsonObject(mapping)); + } + return QJsonObject{ + {QStringLiteral("deviceId"), profile.deviceId}, + {QStringLiteral("protocol"), profile.protocol}, + {QStringLiteral("mappings"), mappings}, + }; +} + } // namespace bool ProfileRegistry::loadFromFile(const QString& filePath) @@ -86,6 +155,26 @@ const softbus::core::models::DeviceDataProfile* ProfileRegistry::find(const QStr return nullptr; } +QJsonArray ProfileRegistry::list() const +{ + QJsonArray out; + for (auto it = m_profilesByKey.cbegin(); it != m_profilesByKey.cend(); ++it) { + out.push_back(profileToJsonObject(it.value())); + } + return out; +} + +QJsonArray ProfileRegistry::listByDeviceId(const QString& deviceId) const +{ + QJsonArray out; + for (auto it = m_profilesByKey.cbegin(); it != m_profilesByKey.cend(); ++it) { + if (it.value().deviceId == deviceId) { + out.push_back(profileToJsonObject(it.value())); + } + } + return out; +} + QString ProfileRegistry::makeKey(const QString& deviceId, const QString& protocol) { return deviceId + QStringLiteral("::") + protocol; diff --git a/src/core/metadata/ProfileRegistry.h b/src/core/metadata/ProfileRegistry.h index 6dc0921..5f0154a 100644 --- a/src/core/metadata/ProfileRegistry.h +++ b/src/core/metadata/ProfileRegistry.h @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -17,6 +18,8 @@ public: bool loadFromFile(const QString& filePath); const softbus::core::models::DeviceDataProfile* find(const QString& deviceId, const QString& protocol) const; + QJsonArray list() const; + QJsonArray listByDeviceId(const QString& deviceId) const; const QString& lastError() const { return m_lastError; } bool empty() const { return m_profilesByKey.isEmpty(); } diff --git a/src/core/plugin_system/PluginManager.cpp b/src/core/plugin_system/PluginManager.cpp index 2c415ff..1fd0120 100644 --- a/src/core/plugin_system/PluginManager.cpp +++ b/src/core/plugin_system/PluginManager.cpp @@ -3,6 +3,44 @@ namespace softbus::core::plugin_system { +namespace +{ + +QString protocolTypeToString(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"); + case softbus::core::models::ProtocolType::UNKNOWN: + default: + return QStringLiteral("UNKNOWN"); + } +} + +QJsonObject makePluginObject(const QString& category, + const QString& pluginId, + softbus::core::models::ProtocolType protocol) +{ + return QJsonObject{ + {QStringLiteral("category"), category}, + {QStringLiteral("pluginId"), pluginId}, + {QStringLiteral("protocol"), protocolTypeToString(protocol)}, + {QStringLiteral("protocolValue"), static_cast(protocol)}, + }; +} + +} // namespace + void PluginManager::registerProtocolPlugin(std::shared_ptr plugin) { if (!plugin) { @@ -95,4 +133,93 @@ QStringList PluginManager::registeredProtocolMapperPluginIds() const return m_protocolMapperPluginsById.keys(); } +QJsonArray PluginManager::protocolPluginsSnapshot() const +{ + QJsonArray out; + for (auto it = m_protocolPluginsById.cbegin(); it != m_protocolPluginsById.cend(); ++it) { + if (!it.value()) { + continue; + } + out.push_back(makePluginObject(QStringLiteral("parser"), it.value()->pluginId(), it.value()->protocolType())); + } + return out; +} + +QJsonArray PluginManager::protocolFramerPluginsSnapshot() const +{ + QJsonArray out; + for (auto it = m_protocolFramerPluginsById.cbegin(); it != m_protocolFramerPluginsById.cend(); ++it) { + if (!it.value()) { + continue; + } + out.push_back(makePluginObject(QStringLiteral("framer"), it.value()->pluginId(), it.value()->protocolType())); + } + return out; +} + +QJsonArray PluginManager::protocolMapperPluginsSnapshot() const +{ + QJsonArray out; + for (auto it = m_protocolMapperPluginsById.cbegin(); it != m_protocolMapperPluginsById.cend(); ++it) { + if (!it.value()) { + continue; + } + out.push_back(makePluginObject(QStringLiteral("mapper"), it.value()->pluginId(), it.value()->protocolType())); + } + return out; +} + +QJsonArray PluginManager::protocolCapabilitiesSnapshot() const +{ + QHash caps; + for (auto it = m_protocolPluginsByType.cbegin(); it != m_protocolPluginsByType.cend(); ++it) { + if (!it.value()) { + continue; + } + auto obj = caps.value(it.key(), QJsonObject{ + {QStringLiteral("protocol"), protocolTypeToString(it.key())}, + {QStringLiteral("protocolValue"), static_cast(it.key())}, + {QStringLiteral("parserPluginId"), QString()}, + {QStringLiteral("framerPluginId"), QString()}, + {QStringLiteral("mapperPluginId"), QString()}, + }); + obj.insert(QStringLiteral("parserPluginId"), it.value()->pluginId()); + caps.insert(it.key(), obj); + } + for (auto it = m_protocolFramerPluginsByType.cbegin(); it != m_protocolFramerPluginsByType.cend(); ++it) { + if (!it.value()) { + continue; + } + auto obj = caps.value(it.key(), QJsonObject{ + {QStringLiteral("protocol"), protocolTypeToString(it.key())}, + {QStringLiteral("protocolValue"), static_cast(it.key())}, + {QStringLiteral("parserPluginId"), QString()}, + {QStringLiteral("framerPluginId"), QString()}, + {QStringLiteral("mapperPluginId"), QString()}, + }); + obj.insert(QStringLiteral("framerPluginId"), it.value()->pluginId()); + caps.insert(it.key(), obj); + } + for (auto it = m_protocolMapperPluginsByType.cbegin(); it != m_protocolMapperPluginsByType.cend(); ++it) { + if (!it.value()) { + continue; + } + auto obj = caps.value(it.key(), QJsonObject{ + {QStringLiteral("protocol"), protocolTypeToString(it.key())}, + {QStringLiteral("protocolValue"), static_cast(it.key())}, + {QStringLiteral("parserPluginId"), QString()}, + {QStringLiteral("framerPluginId"), QString()}, + {QStringLiteral("mapperPluginId"), QString()}, + }); + obj.insert(QStringLiteral("mapperPluginId"), it.value()->pluginId()); + caps.insert(it.key(), obj); + } + + QJsonArray out; + for (auto it = caps.cbegin(); it != caps.cend(); ++it) { + out.push_back(it.value()); + } + return out; +} + } // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/PluginManager.h b/src/core/plugin_system/PluginManager.h index d98a4d5..63cf003 100644 --- a/src/core/plugin_system/PluginManager.h +++ b/src/core/plugin_system/PluginManager.h @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -32,6 +33,10 @@ public: QStringList registeredProtocolPluginIds() const; QStringList registeredProtocolFramerPluginIds() const; QStringList registeredProtocolMapperPluginIds() const; + QJsonArray protocolPluginsSnapshot() const; + QJsonArray protocolFramerPluginsSnapshot() const; + QJsonArray protocolMapperPluginsSnapshot() const; + QJsonArray protocolCapabilitiesSnapshot() const; private: QHash> m_protocolPluginsById; diff --git a/src/device_bus/DeviceBus.cpp b/src/device_bus/DeviceBus.cpp index cf726c6..f1ed91e 100644 --- a/src/device_bus/DeviceBus.cpp +++ b/src/device_bus/DeviceBus.cpp @@ -156,6 +156,41 @@ std::shared_ptr DeviceBus::deviceTreeModel() const { return d->treeModel; } +std::vector DeviceBus::listDevices() const { + if (!d || !d->manager) { + return {}; + } + return d->manager->listDevices(); +} + +std::optional DeviceBus::findDeviceById(int id) const { + if (!d || !d->manager) { + return std::nullopt; + } + return d->manager->findDeviceById(id); +} + +bool DeviceBus::reconcileFromTreeModel() { + if (!d || !d->manager) { + return false; + } + return d->manager->reconcileFromTreeModel(); +} + +softbus::message_bus::pipeline::PipelineEngine* DeviceBus::pipelineEngine() const { + if (!d || !d->manager) { + return nullptr; + } + return d->manager->pipelineEngine(); +} + +std::shared_ptr DeviceBus::pluginManager() const { + if (!d || !d->manager) { + return {}; + } + return d->manager->pluginManager(); +} + void DeviceBus::markDeviceTreeDirty() { if (!d) return; d->deviceTreeDirty = true; diff --git a/src/device_bus/DeviceBus.h b/src/device_bus/DeviceBus.h index d27db60..1b7c7ea 100644 --- a/src/device_bus/DeviceBus.h +++ b/src/device_bus/DeviceBus.h @@ -3,8 +3,21 @@ #include #include #include +#include +#include #include "device_bus/tree/DeviceTreeModel.h" +#include "devices/DeviceTypes.h" + +namespace softbus::core::plugin_system +{ +class PluginManager; +} + +namespace softbus::message_bus::pipeline +{ +class PipelineEngine; +} namespace softbus::device_bus { @@ -22,6 +35,11 @@ public: // 共享设备树模型（单一真源） std::shared_ptr deviceTreeModel() const; + std::vector listDevices() const; + std::optional findDeviceById(int id) const; + bool reconcileFromTreeModel(); + softbus::message_bus::pipeline::PipelineEngine* pipelineEngine() const; + std::shared_ptr pluginManager() const; // 注册/更新设备树后调用，确保 shutdown 会落盘 void markDeviceTreeDirty(); diff --git a/src/device_bus/manager/DeviceBusManager.cpp b/src/device_bus/manager/DeviceBusManager.cpp index 6a6f46f..b12bd21 100644 --- a/src/device_bus/manager/DeviceBusManager.cpp +++ b/src/device_bus/manager/DeviceBusManager.cpp @@ -3,18 +3,128 @@ #include #include +#include +#include +#include #include +#include +#include #include // #include +#include "device_bus/DeviceBus.h" #include "message_bus/egress/EgressRouter.h" #include "message_bus/ingress/DriverIngressAdapter.h" #include "message_bus/pipeline/PipelineEngine.h" +#include "message_bus/pipeline/rules/DynamicRuleRegistry.h" #include "core/plugin_system/PluginManager.h" #include "device_bus/manager/SerialDeviceManager.h" #include "devices/DeviceTypes.h" #include "utils/logs/logging.h" +namespace +{ + +softbus::message_bus::pipeline::RuleScope parseRuleScope(const QJsonObject& obj) +{ + softbus::message_bus::pipeline::RuleScope scope; + scope.endpointHash = static_cast(obj.value(QStringLiteral("endpointHash")).toInteger(0)); + scope.deviceId = obj.value(QStringLiteral("deviceId")).toInt(-1); + scope.stableKey = obj.value(QStringLiteral("stableKey")).toString(); + return scope; +} + +QString rulesConfigPath() +{ + const QString daemonConfigPath = softbus::device_bus::DeviceBus::instance().configPath(); + if (!daemonConfigPath.isEmpty()) { + const QFileInfo fi(daemonConfigPath); + return fi.dir().filePath(QStringLiteral("runtime_rules.json")); + } + return QDir::homePath() + QStringLiteral("/softbus/config/runtime_rules.json"); +} + +void loadDynamicRulesFromDedicatedConfig() +{ + auto& reg = softbus::message_bus::pipeline::DynamicRuleRegistry::instance(); + const QString path = rulesConfigPath(); + QFile file(path); + if (!file.exists()) { + const QFileInfo fi(path); + if (!QDir().mkpath(fi.dir().absolutePath())) { + LOG_WARNING() << "DeviceBusManager: failed to create rules config directory:" << fi.dir().absolutePath(); + return; + } + QSaveFile initFile(path); + if (!initFile.open(QIODevice::WriteOnly | QIODevice::Text)) { + LOG_WARNING() << "DeviceBusManager: failed to create rules config file:" << path; + return; + } + const QJsonObject emptyRules{ + {QStringLiteral("filter"), QJsonArray{}}, + {QStringLiteral("transform"), QJsonArray{}}, + }; + const QByteArray initBytes = QJsonDocument(emptyRules).toJson(QJsonDocument::Indented); + if (initFile.write(initBytes) != initBytes.size()) { + LOG_WARNING() << "DeviceBusManager: failed to initialize rules config file:" << path; + initFile.cancelWriting(); + return; + } + if (!initFile.commit()) { + LOG_WARNING() << "DeviceBusManager: failed to commit initialized rules config file:" << path; + return; + } + LOG_INFO() << "DeviceBusManager: created empty rules config file:" << path; + return; + } + if (!file.open(QIODevice::ReadOnly | QIODevice::Text)) { + LOG_WARNING() << "DeviceBusManager: failed to open rules config:" << path; + return; + } + QJsonParseError pe; + const QJsonDocument doc = QJsonDocument::fromJson(file.readAll(), &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + LOG_WARNING() << "DeviceBusManager: invalid rules config json:" << path << "err=" << pe.errorString(); + return; + } + + const QJsonObject rulesObj = doc.object(); + const QJsonArray filterRules = rulesObj.value(QStringLiteral("filter")).toArray(); + const QJsonArray transformRules = rulesObj.value(QStringLiteral("transform")).toArray(); + int loadedFilterCount = 0; + for (const auto& item : filterRules) { + const QJsonObject obj = item.toObject(); + softbus::message_bus::pipeline::FilterRule rule; + rule.id = obj.value(QStringLiteral("id")).toString(); + rule.scope = parseRuleScope(obj.value(QStringLiteral("scope")).toObject()); + rule.condition = obj.value(QStringLiteral("condition")).toObject(); + if (!rule.id.isEmpty()) { + (void)reg.addFilterRule(rule); + ++loadedFilterCount; + } + } + + int loadedTransformCount = 0; + for (const auto& item : transformRules) { + const QJsonObject obj = item.toObject(); + softbus::message_bus::pipeline::TransformRule rule; + rule.id = obj.value(QStringLiteral("id")).toString(); + rule.scope = parseRuleScope(obj.value(QStringLiteral("scope")).toObject()); + rule.scale = obj.value(QStringLiteral("scale")).toDouble(1.0); + rule.offset = obj.value(QStringLiteral("offset")).toDouble(0.0); + rule.unit = obj.value(QStringLiteral("unit")).toString(); + if (!rule.id.isEmpty()) { + (void)reg.addTransformRule(rule); + ++loadedTransformCount; + } + } + LOG_INFO() << "DeviceBusManager: loaded dynamic rules from" << path + << "filter=" << loadedFilterCount + << "transform=" << loadedTransformCount; +} + +} // namespace + DeviceBusManager::~DeviceBusManager() { shutdown(); @@ -97,6 +207,7 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig) m_pipelineEngine->setMemoryPool(m_memoryPool); m_pipelineEngine->loadRuntimeConfigFromJson(rootConfig.value(QStringLiteral("pipeline")).toObject()); m_pipelineEngine->setPluginManager(m_pluginManager); + loadDynamicRulesFromDedicatedConfig(); m_pipelineEngine->start(); } if (!m_ingressAdapter) { @@ -198,6 +309,32 @@ void DeviceBusManager::shutdown() } } +std::vector DeviceBusManager::listDevices() const +{ + if (!m_registry) { + return {}; + } + return m_registry->listDevices(); +} + +std::optional DeviceBusManager::findDeviceById(int id) const +{ + if (!m_registry) { + return std::nullopt; + } + return m_registry->findById(id); +} + +softbus::message_bus::pipeline::PipelineEngine* DeviceBusManager::pipelineEngine() const +{ + return m_pipelineEngine.get(); +} + +std::shared_ptr DeviceBusManager::pluginManager() const +{ + return m_pluginManager; +} + void DeviceBusManager::initializeSerialDevice(const QString& endpoint, const QJsonObject& params) { softbus::device_bus::manager::SerialInitDeps deps; diff --git a/src/device_bus/manager/DeviceBusManager.h b/src/device_bus/manager/DeviceBusManager.h index 0f201dc..9e17353 100644 --- a/src/device_bus/manager/DeviceBusManager.h +++ b/src/device_bus/manager/DeviceBusManager.h @@ -34,6 +34,10 @@ public: bool initialize(const QJsonObject& rootConfig); bool reconcileFromTreeModel(); void shutdown(); + std::vector listDevices() const; + std::optional findDeviceById(int id) const; + softbus::message_bus::pipeline::PipelineEngine* pipelineEngine() const; + std::shared_ptr pluginManager() const; private: void initializeSerialDevice(const QString& endpoint, const QJsonObject& params); diff --git a/src/devices/DeviceTypes.h b/src/devices/DeviceTypes.h index 0b4d289..908d0c9 100644 --- a/src/devices/DeviceTypes.h +++ b/src/devices/DeviceTypes.h @@ -23,6 +23,29 @@ enum class DeviceKind { Distributed, }; +inline QString deviceKindToString(DeviceKind kind) +{ + switch (kind) { + case DeviceKind::Serial: + return QStringLiteral("serial"); + case DeviceKind::Can: + return QStringLiteral("can"); + case DeviceKind::Tcp: + return QStringLiteral("tcp"); + case DeviceKind::Udp: + return QStringLiteral("udp"); + case DeviceKind::Network: + return QStringLiteral("network"); + case DeviceKind::Virtual: + return QStringLiteral("virtual"); + case DeviceKind::Distributed: + return QStringLiteral("distributed"); + case DeviceKind::Unknown: + default: + return QStringLiteral("unknown"); + } +} + struct DeviceInfo { int id{0}; @@ -38,7 +61,7 @@ struct DeviceInfo bool isActive{true}; QDateTime lastSeen; - QString toJson() const + QJsonObject toJsonObject() const { QJsonObject obj; obj[QStringLiteral("id")] = id; @@ -46,6 +69,7 @@ struct DeviceInfo obj[QStringLiteral("name")] = name; obj[QStringLiteral("type")] = type; obj[QStringLiteral("kind")] = static_cast(kind); + obj[QStringLiteral("kindName")] = deviceKindToString(kind); obj[QStringLiteral("address")] = address; obj[QStringLiteral("protocol")] = protocol; obj[QStringLiteral("protocol_detail")] = protocolDetail; @@ -55,7 +79,12 @@ struct DeviceInfo if (lastSeen.isValid()) { obj[QStringLiteral("lastSeen")] = lastSeen.toString(Qt::ISODate); } - const QJsonDocument doc(obj); + return obj; + } + + QString toJson() const + { + const QJsonDocument doc(toJsonObject()); return QString::fromUtf8(doc.toJson(QJsonDocument::Compact)); } }; diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index c8ffbfb..bc385ca 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -5,7 +5,9 @@ #include #include #include + #include "core/models/MessageRoutingUtils.h" +#include "message_bus/pipeline/PipelineEngineApi.h" #include "message_bus/pipeline/discovery/DiscoveryPool.h" #include "message_bus/pipeline/mapping/MappingRegistry.h" #include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h" @@ -287,6 +289,82 @@ void PipelineEngine::setPluginManager(std::shared_ptr(m_counters.in.load(std::memory_order_relaxed))); + out.insert(QStringLiteral("out"), static_cast(m_counters.out.load(std::memory_order_relaxed))); + out.insert(QStringLiteral("drop"), static_cast(m_counters.drop.load(std::memory_order_relaxed))); + out.insert(QStringLiteral("error"), static_cast(m_counters.error.load(std::memory_order_relaxed))); + out.insert(QStringLiteral("mergeDrop"), static_cast(m_counters.mergeDrop.load(std::memory_order_relaxed))); + return out; +} + +QJsonObject PipelineEngine::runtimeConfigJson() const +{ + QJsonObject out; + out.insert(QStringLiteral("parallelPipeline"), m_rtConfig.parallelPipeline); + out.insert(QStringLiteral("parseWorkerCount"), m_rtConfig.parseWorkerCount); + out.insert(QStringLiteral("orderedEgress"), m_rtConfig.orderedEgress); + out.insert(QStringLiteral("framedQueueMax"), static_cast(m_rtConfig.framedQueueMax)); + out.insert(QStringLiteral("maxReorderDepth"), static_cast(m_rtConfig.maxReorderDepth)); + return out; +} + +QJsonObject PipelineEngine::status() const +{ + int framerSessionCount = 0; + int parserSessionCount = 0; + int mapperSessionCount = 0; + int endpointFramerCount = 0; + int trackedEndpointCount = 0; + { + std::lock_guard lk(m_framerSessionMutex); + framerSessionCount = m_framerSessions.size(); + } + { + std::lock_guard lk(m_parserSessionMutex); + parserSessionCount = m_parserSessions.size(); + } + { + std::lock_guard lk(m_mapperSessionMutex); + mapperSessionCount = m_mapperSessions.size(); + } + { + std::lock_guard lk(m_endpointMutex); + endpointFramerCount = m_endpointFramers.size(); + } + { + std::lock_guard lk(m_seqMutex); + trackedEndpointCount = m_nextFrameSeqByEndpoint.size(); + } + + QJsonObject out; + out.insert(QStringLiteral("running"), m_running.load(std::memory_order_acquire)); + out.insert(QStringLiteral("runtimeConfig"), runtimeConfigJson()); + out.insert(QStringLiteral("counters"), countersJson()); + out.insert(QStringLiteral("parseWorkerCount"), static_cast(m_parseThreads.size())); + out.insert(QStringLiteral("framerSessionCount"), framerSessionCount); + out.insert(QStringLiteral("parserSessionCount"), parserSessionCount); + out.insert(QStringLiteral("mapperSessionCount"), mapperSessionCount); + out.insert(QStringLiteral("endpointFramerCount"), endpointFramerCount); + out.insert(QStringLiteral("trackedEndpointCount"), trackedEndpointCount); + out.insert(QStringLiteral("ingressQueueCapacity"), static_cast(m_ingressQ.capacity())); + out.insert(QStringLiteral("egressQueueCapacity"), static_cast(m_egressQ.capacity())); + out.insert(QStringLiteral("hasPluginManager"), m_pluginManager != nullptr); + if (m_pluginManager) { + out.insert(QStringLiteral("parserPlugins"), QJsonArray::fromStringList(m_pluginManager->registeredProtocolPluginIds())); + out.insert(QStringLiteral("framerPlugins"), QJsonArray::fromStringList(m_pluginManager->registeredProtocolFramerPluginIds())); + out.insert(QStringLiteral("mapperPlugins"), QJsonArray::fromStringList(m_pluginManager->registeredProtocolMapperPluginIds())); + } + return out; +} + // 功能 ：确保端点帧解析器 // 参数 ：endpoint 端点 // 返回值 ：无 @@ -494,7 +572,18 @@ bool PipelineEngine::runStages234(const softbus::core::models::RawBusMessage& ct if (!stage4Validator(msg)) { continue; } - (void)applyTransformAndEnrich(ctx, msg); + const auto enriched = applyTransformAndEnrich(ctx, msg); + const QString engineeringValueText = std::visit( + [](const auto& x) { return QString::number(static_cast(x), 'g', 12); }, + enriched.engineeringValue); + LOG_INFO() << "PipelineEngine: enriched message" + << "sourceMessageId=" << enriched.sourceMessageId + << "logicalName=" << enriched.logicalName + << "engineeringValue=" << engineeringValueText + << "unit=" << enriched.unit + << "quality=" << static_cast(enriched.quality) + << "alarm=" << enriched.alarm + << "traceCount=" << static_cast(enriched.traceLog.size()); } return true; } @@ -521,6 +610,7 @@ std::vector PipelineEngine::stage2Parser(cons if (!parserPlugin) { return {}; } + // 创建解析器会话 const SessionKey parserKey{ctx.endpointHash, parserPlugin->pluginId()}; std::shared_ptr parserSession; { diff --git a/src/message_bus/pipeline/PipelineEngine.h b/src/message_bus/pipeline/PipelineEngine.h index 7cad5d2..08ed095 100644 --- a/src/message_bus/pipeline/PipelineEngine.h +++ b/src/message_bus/pipeline/PipelineEngine.h @@ -83,6 +83,10 @@ public: void loadRuntimeConfigFromJson(const QJsonObject& obj); const Counters& counters() const { return m_counters; } + void resetCounters(); + QJsonObject countersJson() const; + QJsonObject runtimeConfigJson() const; + QJsonObject status() const; private: struct MergeItem @@ -129,16 +133,16 @@ private: PipelineRuntimeConfig m_rtConfig; std::shared_ptr m_passthroughFramer; - std::mutex m_framerSessionMutex; + mutable std::mutex m_framerSessionMutex; QHash> m_framerSessions; - std::mutex m_parserSessionMutex; + mutable std::mutex m_parserSessionMutex; QHash> m_parserSessions; - std::mutex m_mapperSessionMutex; + mutable std::mutex m_mapperSessionMutex; QHash> m_mapperSessions; std::unique_ptr m_mappingLookup; std::unique_ptr m_discoverySink; - std::mutex m_seqMutex; + mutable std::mutex m_seqMutex; QHash m_nextFrameSeqByEndpoint; std::unique_ptr> m_framedQ; @@ -146,7 +150,7 @@ private: std::vector m_parseThreads; std::thread m_mergerThread; - std::mutex m_endpointMutex; + mutable std::mutex m_endpointMutex; QHash> m_endpointFramers; std::mutex m_mergeStateMutex; diff --git a/src/message_bus/pipeline/PipelineEngineApi.cpp b/src/message_bus/pipeline/PipelineEngineApi.cpp new file mode 100644 index 0000000..7ee0751 --- /dev/null +++ b/src/message_bus/pipeline/PipelineEngineApi.cpp @@ -0,0 +1,15 @@ +#include "message_bus/pipeline/PipelineEngineApi.h" + +namespace softbus::message_bus::pipeline +{ + +void resetPipelineCounters(PipelineEngine::Counters& counters) +{ + counters.in.store(0, std::memory_order_relaxed); + counters.out.store(0, std::memory_order_relaxed); + counters.drop.store(0, std::memory_order_relaxed); + counters.error.store(0, std::memory_order_relaxed); + counters.mergeDrop.store(0, std::memory_order_relaxed); +} + +} // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/PipelineEngineApi.h b/src/message_bus/pipeline/PipelineEngineApi.h new file mode 100644 index 0000000..7bdd566 --- /dev/null +++ b/src/message_bus/pipeline/PipelineEngineApi.h @@ -0,0 +1,10 @@ +#pragma once + +#include "message_bus/pipeline/PipelineEngine.h" + +namespace softbus::message_bus::pipeline +{ + +void resetPipelineCounters(PipelineEngine::Counters& counters); + +} // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp b/src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp index bf7d5f4..8fd4bcf 100644 --- a/src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp +++ b/src/message_bus/pipeline/rules/DynamicRuleRegistry.cpp @@ -6,6 +6,32 @@ namespace softbus::message_bus::pipeline { +namespace +{ + +// 规则里可能写完整 routingKey 的十进制串（与 routingKeyToStableKey 一致），也可能写 UI 侧常用的 +// "ep_<十进制 endpointHash>"（仅按物理端点匹配，不含 deviceId 位）。 +bool stableKeyMatches(const QString& pattern, + const QString& messageStableKey, + std::uint32_t endpointHash) +{ + if (pattern.isEmpty()) { + return true; + } + if (pattern == messageStableKey) { + return true; + } + if (pattern.startsWith(QStringLiteral("ep_"), Qt::CaseInsensitive)) { + bool ok = false; + const QString digits = pattern.mid(3); + const quint32 parsed = digits.toUInt(&ok, 10); + return ok && parsed == endpointHash; + } + return false; +} + +} // namespace + DynamicRuleRegistry& DynamicRuleRegistry::instance() { static DynamicRuleRegistry registry; @@ -72,6 +98,28 @@ std::vector DynamicRuleRegistry::snapshotTransformRules( return out; } +std::vector DynamicRuleRegistry::snapshotAllFilterRules() const +{ + std::shared_lock lk(m_mtx); + std::vector out; + out.reserve(static_cast(m_filterRules.size())); + for (auto it = m_filterRules.constBegin(); it != m_filterRules.constEnd(); ++it) { + out.push_back(it.value()); + } + return out; +} + +std::vector DynamicRuleRegistry::snapshotAllTransformRules() const +{ + std::shared_lock lk(m_mtx); + std::vector out; + out.reserve(static_cast(m_transformRules.size())); + for (auto it = m_transformRules.constBegin(); it != m_transformRules.constEnd(); ++it) { + out.push_back(it.value()); + } + return out; +} + QJsonObject DynamicRuleRegistry::status() const { std::shared_lock lk(m_mtx); @@ -86,7 +134,7 @@ bool DynamicRuleRegistry::matchScope( { 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; + const bool stableKeyOk = stableKeyMatches(scope.stableKey, stableKey, endpointHash); return endpointHashOk && deviceOk && stableKeyOk; } diff --git a/src/message_bus/pipeline/rules/DynamicRuleRegistry.h b/src/message_bus/pipeline/rules/DynamicRuleRegistry.h index b74c3dc..623f7ea 100644 --- a/src/message_bus/pipeline/rules/DynamicRuleRegistry.h +++ b/src/message_bus/pipeline/rules/DynamicRuleRegistry.h @@ -46,6 +46,8 @@ public: std::vector snapshotFilterRules(std::uint32_t endpointHash, int deviceId, const QString& stableKey) const; std::vector snapshotTransformRules(std::uint32_t endpointHash, int deviceId, const QString& stableKey) const; + std::vector snapshotAllFilterRules() const; + std::vector snapshotAllTransformRules() const; QJsonObject status() const;