diff --git a/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx b/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx index b685fc0..966eddf 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 1bbe327..e8a76d3 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/IpcDBus.cpp.F75762A8BF6179E0.idx b/.cache/clangd/index/IpcDBus.cpp.F75762A8BF6179E0.idx index 824d38f..83a7e4f 100644 Binary files a/.cache/clangd/index/IpcDBus.cpp.F75762A8BF6179E0.idx and b/.cache/clangd/index/IpcDBus.cpp.F75762A8BF6179E0.idx differ diff --git a/.cache/clangd/index/IpcDBus.h.DD48681835A8EA94.idx b/.cache/clangd/index/IpcDBus.h.DD48681835A8EA94.idx index 2f1cab4..a86660a 100644 Binary files a/.cache/clangd/index/IpcDBus.h.DD48681835A8EA94.idx and b/.cache/clangd/index/IpcDBus.h.DD48681835A8EA94.idx differ diff --git a/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx b/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx index 5314e6c..d10ad8f 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 d0f9e8e..f29aca5 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -60,6 +60,8 @@ qt_add_executable(softbus_daemon src/api/ipc_dbus/IpcDBus.h src/api/ipc_dbus/IpcDBus.cpp + src/api/ipc_dbus/CommandDispatcher.h + src/api/ipc_dbus/CommandDispatcher.cpp # device types src/devices/DeviceTypes.h @@ -118,16 +120,20 @@ qt_add_executable(softbus_daemon src/core/memory/LockFreeQueue.h src/core/plugin_system/IProtocolPlugin.h src/core/plugin_system/IProtocolFramerPlugin.h + src/core/plugin_system/IProtocolMapperPlugin.h src/core/plugin_system/PluginManager.h src/core/plugin_system/PluginManager.cpp src/core/plugin_system/PluginHost.h src/core/plugin_system/PluginHost.cpp - plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h - plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.cpp + plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h + plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp + plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h + plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp src/core/models/Types.h src/core/models/MetadataDef.h src/core/models/DataMapping.h src/core/models/DOMMessage.h + src/core/models/EnrichedMessage.h src/core/metadata/MetadataRegistry.h src/core/metadata/MetadataRegistry.cpp src/core/metadata/ProfileRegistry.h @@ -136,16 +142,18 @@ qt_add_executable(softbus_daemon # message bus pipeline src/message_bus/pipeline/PayloadRef.h src/message_bus/pipeline/PipelineContext.h + src/message_bus/pipeline/DynamicRuleRegistry.h + src/message_bus/pipeline/DynamicRuleRegistry.cpp src/message_bus/pipeline/PipelineEngine.h src/message_bus/pipeline/PipelineEngine.cpp src/message_bus/pipeline/protocol/ProtocolEnvelope.h - plugins/protocols/modbusrtu/types/ModbusRtuPdu.h - plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h - plugins/protocols/modbusrtu/codec/ModbusRtuCodec.cpp - plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h - plugins/protocols/modbusrtu/framer/ModbusRtuFraming.cpp - plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h - plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.cpp + plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h + plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h + plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp + plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h + plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.cpp + plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h + plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp src/message_bus/pipeline/stages/1_framers/IFramer.h src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h diff --git a/bin/softbus_daemon b/bin/softbus_daemon index 3a3d729..4131c63 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/bin/softbus_modbus_rtu_framer_test b/bin/softbus_modbus_rtu_framer_test deleted file mode 100644 index f4a3446..0000000 Binary files a/bin/softbus_modbus_rtu_framer_test and /dev/null differ diff --git a/bin/softbus_modbus_rtu_framing_predict_test b/bin/softbus_modbus_rtu_framing_predict_test deleted file mode 100644 index c758ea8..0000000 Binary files a/bin/softbus_modbus_rtu_framing_predict_test and /dev/null differ diff --git a/bin/softbus_modbus_rtu_pdu_codec_test b/bin/softbus_modbus_rtu_pdu_codec_test deleted file mode 100644 index 6f144b6..0000000 Binary files a/bin/softbus_modbus_rtu_pdu_codec_test and /dev/null differ diff --git a/bin/softbus_protocol_parse_filter_test b/bin/softbus_protocol_parse_filter_test deleted file mode 100644 index d8c1409..0000000 Binary files a/bin/softbus_protocol_parse_filter_test and /dev/null differ diff --git a/config/daemon_config.json b/config/daemon_config.json deleted file mode 100644 index fdb63de..0000000 --- a/config/daemon_config.json +++ /dev/null @@ -1,46 +0,0 @@ -{ - "plugin_dir": "bin/plugins", - "pipeline": { - "parallelPipeline": false, - "parseWorkerCount": 2, - "orderedEgress": true, - "framedQueueMax": 4096, - "maxReorderDepth": 256, - "modbusRtuMaxFrameBytes": 256, - "modbusRtuInterFrameTimeoutMs": 0 - }, - "plugins": { - "framers": [ - { - "path": "bin/plugins/libmodbus_rtu.so", - "protocol": "modbus_rtu", - "abi": 1 - } - ], - "hotReload": { - "enabled": true, - "pollIntervalMs": 2000 - } - }, - "device_tree": [ - { - "id": "serial_1", - "name": "Serial Device 1", - "type": "serial", - "endpoint": "/dev/ttyS1", - "driver": "hardware/drivers/serial/SerialQtDriver", - "capabilitiesRef": "src/hardware/drivers/serial/parameters.json", - "params": { - "protocol": "modbus_rtu", - "baudRate": 115200, - "dataBits": 8, - "parity": "none", - "stopBits": 1, - "flowControl": "none" - }, - "children": [], - "enabled": true - } - ], - "routes": [] -} diff --git a/config/device_profiles.json b/config/device_profiles.json deleted file mode 100644 index 740947c..0000000 --- a/config/device_profiles.json +++ /dev/null @@ -1,36 +0,0 @@ -{ - "profiles": [ - { - "deviceId": "*", - "protocol": "modbus_rtu", - "mappings": [ - { - "pointId": "temperature", - "metadataId": "MD_TEMP_001", - "registerAddress": 40001, - "byteOffset": 0, - "byteLength": 4, - "bitOffset": -1, - "endianness": "BIG_ENDIAN", - "rawType": "INT32", - "scaleFactor": 1.0, - "offsetFactor": 0.0, - "expression": "x * 0.1" - }, - { - "pointId": "pressure", - "metadataId": "MD_PRESS_001", - "registerAddress": 40003, - "byteOffset": 4, - "byteLength": 4, - "bitOffset": -1, - "endianness": "BIG_ENDIAN", - "rawType": "INT32", - "scaleFactor": 1.0, - "offsetFactor": 0.0, - "expression": "x * 0.001" - } - ] - } - ] -} diff --git a/config/metadata_dictionary.json b/config/metadata_dictionary.json deleted file mode 100644 index 9f7d2ac..0000000 --- a/config/metadata_dictionary.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "metadatas": [ - { - "metadataId": "MD_TEMP_001", - "name": "主管道温度", - "description": "由两个寄存器组合得到温度", - "domain": "PROCESS_PARAM", - "semanticType": "FLOAT64", - "unit": "C", - "minValue": -40.0, - "maxValue": 180.0, - "deadband": 0.1, - "isWriteable": false - }, - { - "metadataId": "MD_PRESS_001", - "name": "主管道压力", - "description": "由两个寄存器组合得到压力", - "domain": "PROCESS_PARAM", - "semanticType": "FLOAT64", - "unit": "MPa", - "minValue": 0.0, - "maxValue": 25.0, - "deadband": 0.01, - "isWriteable": false - } - ] -} diff --git a/config/plugins_config.json b/config/plugins_config.json deleted file mode 100644 index 4126bc7..0000000 --- a/config/plugins_config.json +++ /dev/null @@ -1,5 +0,0 @@ -{ - "modbus_rtu": { - "register_map": [] - } -} diff --git a/plugins/protocols/modbus_rtu/CMakeLists.txt b/plugins/protocols/modbus_rtu/CMakeLists.txt index e97ef21..5c3fea9 100644 --- a/plugins/protocols/modbus_rtu/CMakeLists.txt +++ b/plugins/protocols/modbus_rtu/CMakeLists.txt @@ -1,10 +1,11 @@ # Runtime-loadable Modbus RTU parser + framer plugin add_library(modbus_rtu SHARED plugin_stub.cpp - ../modbusrtu/parser/ModbusRtuProtocolPlugin.cpp - ../modbusrtu/codec/ModbusRtuCodec.cpp - ../modbusrtu/framer/ModbusRtuFramerPlugin.cpp - ../modbusrtu/framer/ModbusRtuFraming.cpp + parser/ModbusRtuProtocolPlugin.cpp + mapper/ModbusRtuProtocolMapperPlugin.cpp + codec/ModbusRtuCodec.cpp + framer/ModbusRtuFramerPlugin.cpp + framer/ModbusRtuFraming.cpp ) target_include_directories(modbus_rtu PRIVATE ${CMAKE_SOURCE_DIR} diff --git a/plugins/protocols/modbusrtu/README.md b/plugins/protocols/modbus_rtu/README.md similarity index 95% rename from plugins/protocols/modbusrtu/README.md rename to plugins/protocols/modbus_rtu/README.md index 262ce7b..1b6ec82 100644 --- a/plugins/protocols/modbusrtu/README.md +++ b/plugins/protocols/modbus_rtu/README.md @@ -1,4 +1,4 @@ -# Modbus RTU 协议模块（`modbusrtu`） +# Modbus RTU 协议模块（`modbus_rtu`） 本目录实现 **Modbus RTU** 的成帧、编解码与解析插件，供守护进程静态编译与 `plugins/protocols/modbus_rtu` 动态库（`.so`）共用同一套源码。 @@ -26,7 +26,7 @@ parser （依赖 codec、types；不反向依赖 framer 插件类） ## 与动态库的关系 - 动态库入口：`plugins/protocols/modbus_rtu/plugin_stub.cpp`，导出 `softbus_create_protocol_plugin` / `softbus_create_framer_plugin` 等符号。 -- 主工程 `CMakeLists.txt` 同样列出本目录下源文件，用于无 `.so` 时的内置链接；路径与 include 均以 `plugins/protocols/modbusrtu/...` 为准。 +- 主工程 `CMakeLists.txt` 同样列出本目录下源文件，用于无 `.so` 时的内置链接；路径与 include 均以 `plugins/protocols/modbus_rtu/...` 为准。 ## 配置提示 diff --git a/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.cpp b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp similarity index 99% rename from plugins/protocols/modbusrtu/codec/ModbusRtuCodec.cpp rename to plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp index 313eb56..4961d7e 100644 --- a/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.cpp +++ b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp @@ -1,9 +1,9 @@ -#include "plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h" +#include "plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h" #include #include "message_bus/pipeline/PipelineContext.h" -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" namespace softbus::message_bus::pipeline::protocol { diff --git a/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h similarity index 92% rename from plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h rename to plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h index 6534557..0be068c 100644 --- a/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h +++ b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h @@ -6,7 +6,7 @@ #include #include -#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" +#include "plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h" namespace softbus::message_bus::pipeline { diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.cpp b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp similarity index 96% rename from plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.cpp rename to plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp index f639d05..d9b4816 100644 --- a/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.cpp +++ b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp @@ -1,9 +1,9 @@ -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h" #include #include -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" #include "utils/logs/logging.h" namespace softbus::message_bus::pipeline::protocol::modbusrtu @@ -72,7 +72,6 @@ public: auto blk = m_pool->allocate(*pred); if (!blk || !blk->data || blk->capacity < *pred) { - in.framerError = QStringLiteral("pool_exhausted"); return; } std::memcpy(blk->data, m_buf.data(), *pred); diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h b/plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h similarity index 100% rename from plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h rename to plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.cpp b/plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.cpp similarity index 97% rename from plugins/protocols/modbusrtu/framer/ModbusRtuFraming.cpp rename to plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.cpp index b827d02..9cfe4e1 100644 --- a/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.cpp +++ b/plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.cpp @@ -1,4 +1,4 @@ -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" namespace softbus::message_bus::pipeline::protocol::modbusrtu { diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h b/plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h similarity index 100% rename from plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h rename to plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h diff --git a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp new file mode 100644 index 0000000..5733338 --- /dev/null +++ b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.cpp @@ -0,0 +1,93 @@ +#include "plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h" + +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" + +namespace softbus::message_bus::pipeline::protocol::modbusrtu +{ + +namespace +{ + +class ModbusRtuMapperSession final : public softbus::core::plugin_system::IProtocolMapperSession +{ +public: + std::vector map( + const softbus::message_bus::pipeline::PipelineContext& rawContext, + const softbus::message_bus::pipeline::protocol::ProtocolEnvelope& envelope) override + { + std::vector out; + if (envelope.family != softbus::message_bus::pipeline::protocol::ProtocolFamily::ModbusRtu) { + return out; + } + const auto* pdu = std::get_if(&envelope.pdu); + if (!pdu) { + return out; + } + + if (const auto* resp = + std::get_if(pdu)) { + out.reserve(static_cast(resp->registers.size())); + for (int i = 0; i < resp->registers.size(); ++i) { + softbus::core::models::DOMMessage msg; + msg.messageId = QStringLiteral("%1.reg.%2") + .arg(rawContext.stableKey.isEmpty() ? QString::number(rawContext.deviceId) + : rawContext.stableKey) + .arg(i); + msg.timestamp = rawContext.timestamp.toMSecsSinceEpoch(); + msg.value = static_cast(resp->registers[static_cast(i)]); + msg.quality = softbus::core::models::DataQuality::GOOD; + msg.attributes.insert(QStringLiteral("sourceEndpoint"), rawContext.endpoint); + out.push_back(std::move(msg)); + } + return out; + } + + if (const auto* req = + std::get_if(pdu)) { + softbus::core::models::DOMMessage msg; + msg.messageId = QStringLiteral("%1.req.read_holding") + .arg(rawContext.stableKey.isEmpty() ? QString::number(rawContext.deviceId) + : rawContext.stableKey); + msg.timestamp = rawContext.timestamp.toMSecsSinceEpoch(); + msg.value = static_cast(req->quantity); + msg.quality = softbus::core::models::DataQuality::GOOD; + msg.attributes.insert(QStringLiteral("sourceEndpoint"), rawContext.endpoint); + 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)); + } + return out; + } + + void reset() override {} +}; + +class ModbusRtuProtocolMapperPlugin final : public softbus::core::plugin_system::IProtocolMapperPlugin +{ +public: + QString pluginId() const override { return QStringLiteral("modbus_rtu_mapper"); } + bool supports(const QString& protocolHint) const override + { + return isModbusRtuHint(protocolHint); + } + std::unique_ptr createMapperSession() override + { + return std::make_unique(); + } +}; + +} // namespace + +std::shared_ptr makeModbusRtuProtocolMapperPlugin() +{ + return std::make_shared(); +} + +softbus::core::plugin_system::IProtocolMapperPlugin* createModbusRtuProtocolMapperPluginRaw() +{ + return new ModbusRtuProtocolMapperPlugin(); +} + +} // namespace softbus::message_bus::pipeline::protocol::modbusrtu + diff --git a/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h new file mode 100644 index 0000000..52d4c3a --- /dev/null +++ b/plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h @@ -0,0 +1,14 @@ +#pragma once + +#include + +#include "core/plugin_system/IProtocolMapperPlugin.h" + +namespace softbus::message_bus::pipeline::protocol::modbusrtu +{ + +std::shared_ptr makeModbusRtuProtocolMapperPlugin(); +softbus::core::plugin_system::IProtocolMapperPlugin* createModbusRtuProtocolMapperPluginRaw(); + +} // namespace softbus::message_bus::pipeline::protocol::modbusrtu + diff --git a/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp new file mode 100644 index 0000000..658d765 --- /dev/null +++ b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp @@ -0,0 +1,107 @@ +#include "plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h" + +#include "core/plugin_system/IProtocolPlugin.h" +#include "message_bus/pipeline/PipelineContext.h" +#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" +#include "plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" + +#include +#include +#include + +namespace softbus::core::plugin_system +{ + +namespace +{ + +using softbus::message_bus::pipeline::protocol::ModbusExceptionPdu; +using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; +using softbus::message_bus::pipeline::protocol::ModbusRtuUnsupportedPdu; +using softbus::message_bus::pipeline::protocol::ProtocolFamily; +using softbus::message_bus::pipeline::protocol::ReadCoilsRequest; +using softbus::message_bus::pipeline::protocol::ReadCoilsResponse; +using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsRequest; +using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsResponse; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersRequest; +using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; +using softbus::message_bus::pipeline::protocol::ReadInputRegistersRequest; +using softbus::message_bus::pipeline::protocol::ReadInputRegistersResponse; +using softbus::message_bus::pipeline::protocol::WriteSingleCoilRequest; +using softbus::message_bus::pipeline::protocol::WriteSingleCoilResponse; +using softbus::message_bus::pipeline::protocol::WriteSingleRegisterRequest; +using softbus::message_bus::pipeline::protocol::WriteSingleRegisterResponse; +using softbus::message_bus::pipeline::protocol::WriteMultipleCoilsRequest; +using softbus::message_bus::pipeline::protocol::WriteMultipleCoilsResponse; +using softbus::message_bus::pipeline::protocol::WriteMultipleRegistersRequest; +using softbus::message_bus::pipeline::protocol::WriteMultipleRegistersResponse; +using softbus::message_bus::pipeline::protocol::decodeModbusRtu; + +template +struct overloaded : Ts... +{ + using Ts::operator()...; +}; +template +overloaded(Ts...) -> overloaded; + +class ModbusRtuSession final : public IProtocolSession +{ +public: + bool parse(const softbus::message_bus::pipeline::PipelineContext& ctx, + softbus::message_bus::pipeline::protocol::ProtocolEnvelope& out) override + { + using softbus::message_bus::pipeline::FrameKind; + if (ctx.frameKind != FrameKind::CompleteFrame || !ctx.payload.valid()) { + return false; + } + const std::size_t n = ctx.payloadSize ? ctx.payloadSize : ctx.payload.length; + const std::uint8_t* p = ctx.payload.bytes(); + if (!p || n < 4) { + return false; + } + + ModbusRtuPdu pdu; + if (!decodeModbusRtu(ctx.direction, p, n, pdu, nullptr)) { + out.family = ProtocolFamily::None; + out.pdu = {}; + return false; + } + out.family = ProtocolFamily::ModbusRtu; + out.pdu = pdu; + return true; + } + + void reset() override {} +}; + +} // namespace + +class ModbusRtuProtocolPlugin final : public IProtocolPlugin +{ +public: + QString pluginId() const override { return QStringLiteral("modbus_rtu"); } + + bool supports(const QString& protocolHint) const override + { + return softbus::message_bus::pipeline::protocol::modbusrtu::isModbusRtuHint(protocolHint); + } + + std::unique_ptr createSession() override + { + return std::make_unique(); + } +}; + +std::shared_ptr makeModbusRtuProtocolPlugin() +{ + return std::make_shared(); +} + +IProtocolPlugin* createModbusRtuProtocolPluginRaw() +{ + return new ModbusRtuProtocolPlugin(); +} + +} // namespace softbus::core::plugin_system diff --git a/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h similarity index 88% rename from plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h rename to plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h index 673e98f..f5325a7 100644 --- a/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h +++ b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h @@ -1,5 +1,5 @@ #pragma once - +// Modbus RTU 协议解析器插件 #include #include "core/plugin_system/IProtocolPlugin.h" diff --git a/plugins/protocols/modbus_rtu/plugin_stub.cpp b/plugins/protocols/modbus_rtu/plugin_stub.cpp index 30f3acb..a6c6638 100644 --- a/plugins/protocols/modbus_rtu/plugin_stub.cpp +++ b/plugins/protocols/modbus_rtu/plugin_stub.cpp @@ -1,5 +1,6 @@ -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h" -#include "plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.h" +#include "plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h" +#include "plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h" extern "C" int softbus_plugin_abi_version() { @@ -15,3 +16,8 @@ extern "C" softbus::core::plugin_system::IProtocolFramerPlugin* softbus_create_f { return softbus::message_bus::pipeline::protocol::modbusrtu::createModbusRtuFramerPluginRaw(); } + +extern "C" softbus::core::plugin_system::IProtocolMapperPlugin* softbus_create_mapper_plugin() +{ + return softbus::message_bus::pipeline::protocol::modbusrtu::createModbusRtuProtocolMapperPluginRaw(); +} diff --git a/plugins/protocols/modbusrtu/types/ModbusRtuPdu.h b/plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h similarity index 100% rename from plugins/protocols/modbusrtu/types/ModbusRtuPdu.h rename to plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h diff --git a/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.cpp b/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.cpp deleted file mode 100644 index 2c69fda..0000000 --- a/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.cpp +++ /dev/null @@ -1,249 +0,0 @@ -#include "plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h" - -#include "core/plugin_system/IProtocolPlugin.h" -#include "message_bus/pipeline/PipelineContext.h" -#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" -#include "plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h" -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" - -#include -#include - -#include -#include -#include - -namespace softbus::core::plugin_system -{ - -namespace -{ - -using softbus::message_bus::pipeline::protocol::ModbusExceptionPdu; -using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; -using softbus::message_bus::pipeline::protocol::ModbusRtuUnsupportedPdu; -using softbus::message_bus::pipeline::protocol::ProtocolFamily; -using softbus::message_bus::pipeline::protocol::ReadCoilsRequest; -using softbus::message_bus::pipeline::protocol::ReadCoilsResponse; -using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsRequest; -using softbus::message_bus::pipeline::protocol::ReadDiscreteInputsResponse; -using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersRequest; -using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; -using softbus::message_bus::pipeline::protocol::ReadInputRegistersRequest; -using softbus::message_bus::pipeline::protocol::ReadInputRegistersResponse; -using softbus::message_bus::pipeline::protocol::WriteSingleCoilRequest; -using softbus::message_bus::pipeline::protocol::WriteSingleCoilResponse; -using softbus::message_bus::pipeline::protocol::WriteSingleRegisterRequest; -using softbus::message_bus::pipeline::protocol::WriteSingleRegisterResponse; -using softbus::message_bus::pipeline::protocol::WriteMultipleCoilsRequest; -using softbus::message_bus::pipeline::protocol::WriteMultipleCoilsResponse; -using softbus::message_bus::pipeline::protocol::WriteMultipleRegistersRequest; -using softbus::message_bus::pipeline::protocol::WriteMultipleRegistersResponse; -using softbus::message_bus::pipeline::protocol::decodeModbusRtu; - -template -struct overloaded : Ts... -{ - using Ts::operator()...; -}; -template -overloaded(Ts...) -> overloaded; - -void syncLegacyJson(const ModbusRtuPdu& pdu, QJsonObject& o) -{ - o = QJsonObject(); - std::visit( - overloaded{ - [&](const ModbusExceptionPdu& e) { - o.insert(QStringLiteral("unitId"), e.unitId); - o.insert(QStringLiteral("function"), e.function); - o.insert(QStringLiteral("exceptionCode"), e.exceptionCode); - o.insert(QStringLiteral("exception"), true); - }, - [&](const ReadHoldingRegistersResponse& r) { - QJsonArray regs; - for (std::uint16_t v : r.registers) { - regs.append(static_cast(v)); - } - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 3); - o.insert(QStringLiteral("byteCount"), r.registers.size() * 2); - o.insert(QStringLiteral("registers"), regs); - }, - [&](const ReadHoldingRegistersRequest& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 3); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - }, - [&](const ReadCoilsRequest& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 1); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - }, - [&](const ReadDiscreteInputsRequest& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 2); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - }, - [&](const ReadInputRegistersRequest& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 4); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - }, - [&](const ReadCoilsResponse& r) { - QJsonArray bits; - for (std::uint8_t b : r.coils) { - bits.append(static_cast(b)); - } - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 1); - o.insert(QStringLiteral("byteCount"), r.coils.size()); - o.insert(QStringLiteral("coils"), bits); - }, - [&](const ReadDiscreteInputsResponse& r) { - QJsonArray bits; - for (std::uint8_t b : r.discreteInputs) { - bits.append(static_cast(b)); - } - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 2); - o.insert(QStringLiteral("byteCount"), r.discreteInputs.size()); - o.insert(QStringLiteral("discreteInputs"), bits); - }, - [&](const ReadInputRegistersResponse& r) { - QJsonArray regs; - for (std::uint16_t v : r.inputRegisters) { - regs.append(static_cast(v)); - } - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 4); - o.insert(QStringLiteral("byteCount"), r.inputRegisters.size() * 2); - o.insert(QStringLiteral("inputRegisters"), regs); - }, - [&](const WriteSingleCoilRequest& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 5); - o.insert(QStringLiteral("address"), r.address); - o.insert(QStringLiteral("value"), r.value); - }, - [&](const WriteSingleCoilResponse& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 5); - o.insert(QStringLiteral("address"), r.address); - o.insert(QStringLiteral("value"), r.value); - }, - [&](const WriteSingleRegisterRequest& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 6); - o.insert(QStringLiteral("address"), r.address); - o.insert(QStringLiteral("value"), r.value); - }, - [&](const WriteSingleRegisterResponse& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 6); - o.insert(QStringLiteral("address"), r.address); - o.insert(QStringLiteral("value"), r.value); - }, - [&](const WriteMultipleCoilsRequest& r) { - QJsonArray vals; - for (std::uint8_t b : r.values) { - vals.append(static_cast(b)); - } - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 15); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - o.insert(QStringLiteral("values"), vals); - }, - [&](const WriteMultipleCoilsResponse& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 15); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - }, - [&](const WriteMultipleRegistersRequest& r) { - QJsonArray vals; - for (std::uint16_t v : r.values) { - vals.append(static_cast(v)); - } - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 16); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - o.insert(QStringLiteral("values"), vals); - }, - [&](const WriteMultipleRegistersResponse& r) { - o.insert(QStringLiteral("unitId"), r.unitId); - o.insert(QStringLiteral("function"), 16); - o.insert(QStringLiteral("startAddress"), r.startAddress); - o.insert(QStringLiteral("quantity"), r.quantity); - }, - [&](const ModbusRtuUnsupportedPdu&) {}, - [&](const auto&) {}}, - pdu); -} - -class ModbusRtuSession final : public IProtocolSession -{ -public: - bool feed(softbus::message_bus::pipeline::PipelineContext& ctx) override - { - using softbus::message_bus::pipeline::FrameKind; - if (ctx.frameKind != FrameKind::CompleteFrame || !ctx.payload.valid()) { - return false; - } - const std::size_t n = ctx.payloadSize ? ctx.payloadSize : ctx.payload.length; - const std::uint8_t* p = ctx.payload.bytes(); - if (!p || n < 4) { - return false; - } - - ModbusRtuPdu pdu; - if (!decodeModbusRtu(ctx.direction, p, n, pdu, nullptr)) { - ctx.protocol.family = ProtocolFamily::None; - ctx.protocol.pdu = {}; - ctx.parsed = QJsonObject(); - return false; - } - ctx.protocol.family = ProtocolFamily::ModbusRtu; - ctx.protocol.pdu = pdu; - syncLegacyJson(pdu, ctx.parsed); - return true; - } - - void reset() override {} -}; - -} // namespace - -class ModbusRtuProtocolPlugin final : public IProtocolPlugin -{ -public: - QString pluginId() const override { return QStringLiteral("modbus_rtu"); } - - bool supports(const QString& protocolHint) const override - { - return softbus::message_bus::pipeline::protocol::modbusrtu::isModbusRtuHint(protocolHint); - } - - std::unique_ptr createSession() override - { - return std::make_unique(); - } -}; - -std::shared_ptr makeModbusRtuProtocolPlugin() -{ - return std::make_shared(); -} - -IProtocolPlugin* createModbusRtuProtocolPluginRaw() -{ - return new ModbusRtuProtocolPlugin(); -} - -} // namespace softbus::core::plugin_system diff --git a/src/api/API_INTERFACE_TABLE.md b/src/api/API_INTERFACE_TABLE.md new file mode 100644 index 0000000..c8d6289 --- /dev/null +++ b/src/api/API_INTERFACE_TABLE.md @@ -0,0 +1,83 @@ +# softbus_daemon API 接口表 + +本文档汇总 `src/api/ipc_dbus` 当前对外接口，分为： +- D-Bus 直接方法 +- `executeCommand` JSON 命令接口 + +## 1) D-Bus 服务基础信息 + +| 项目 | 值 | +|---|---| +| Service | `com.softbus.Daemon` | +| ObjectPath | `/com/softbus/Daemon` | +| Interface | `com.softbus.Daemon` | +| 适配类 | `IpcDBus` | + +## 2) D-Bus 方法表（IpcDBus） + +| 方法 | 入参 | 返回 | 说明 | +|---|---|---|---| +| `isRunning()` | 无 | `bool` | 核心服务是否初始化完成 | +| `ping()` | 无 | `QString` | 健康探针，固定返回 `pong` | +| `getVersion()` | 无 | `QString` | 进程版本号 | +| `getStatus()` | 无 | `QString(JSON)` | 返回服务状态（pid/role/running/timestamp 等） | +| `getPid()` | 无 | `qint64` | 返回守护进程 PID | +| `shutdown()` | 无 | `void` | 关闭 CoreService 并退出进程 | +| `reloadConfig()` | 无 | `bool` | 预留，当前为占位实现 | +| `executeCommand(requestJson)` | `QString(JSON)` | `QString(JSON)` | 统一 JSON 命令入口 | + +## 3) executeCommand 协议 + +### 3.1 请求格式 + +```json +{ + "id": "req-001", + "action": "add_filter", + "params": {} +} +``` + +### 3.2 响应格式 + +```json +{ + "id": "req-001", + "code": 0, + "message": "ok", + "data": {} +} +``` + +## 4) action 路由表（CommandDispatcher） + +| action | params 关键字段 | 成功 code | 失败 code | 说明 | +|---|---|---:|---:|---| +| `add_filter` | `id`, `endpoint?`, `deviceId?`, `stableKey?`, `condition?` | `0` | `4001` | 添加过滤规则 | +| `remove_filter` | `id` | `0` | `4041` | 删除过滤规则 | +| `add_transform_rule` | `id`, `endpoint?`, `deviceId?`, `stableKey?`, `scale?`, `offset?`, `unit?` | `0` | `4002` | 添加转换规则 | +| `remove_transform_rule` | `id` | `0` | `4042` | 删除转换规则 | +| `list_rules` | 无 | `0` | - | 返回规则计数状态 | +| `get_status` | 无 | `0` | - | 返回规则计数状态 | +| 其他未知 action | - | - | `4040` | 未知动作 | + +## 5) 通用错误码约定（当前实现） + +| code | message | 说明 | +|---:|---|---| +| `0` | `ok` | 成功 | +| `4000` | `invalid_json` | 请求 JSON 无法解析或非对象 | +| `4001` | `invalid_filter_rule` | 过滤规则参数非法 | +| `4002` | `invalid_transform_rule` | 转换规则参数非法 | +| `4003` | `invalid_action` | action 为空或分发器不可用 | +| `4040` | `unknown_action` | action 未注册 | +| `4041` | `filter_not_found` | 删除过滤规则时未找到 | +| `4042` | `transform_rule_not_found` | 删除转换规则时未找到 | + +## 6) 维护建议 + +- 新增 action 时，同时更新： + - `CommandDispatcher::dispatch(...)` + - 本文档第 4 节 action 表 +- 若扩展返回结构，保持 `code/message/data` 三元组不变，避免客户端兼容性问题。 + diff --git a/src/api/ipc_dbus/CommandDispatcher.cpp b/src/api/ipc_dbus/CommandDispatcher.cpp new file mode 100644 index 0000000..d6b8da8 --- /dev/null +++ b/src/api/ipc_dbus/CommandDispatcher.cpp @@ -0,0 +1,66 @@ +#include "api/ipc_dbus/CommandDispatcher.h" + +#include "message_bus/pipeline/DynamicRuleRegistry.h" + +namespace softbus::api::ipc +{ + +namespace +{ + +softbus::message_bus::pipeline::RuleScope parseScope(const QJsonObject& params) +{ + softbus::message_bus::pipeline::RuleScope scope; + scope.endpoint = params.value(QStringLiteral("endpoint")).toString(); + scope.deviceId = params.value(QStringLiteral("deviceId")).toInt(-1); + scope.stableKey = params.value(QStringLiteral("stableKey")).toString(); + return scope; +} + +} // namespace + +DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObject& params) const +{ + auto& reg = softbus::message_bus::pipeline::DynamicRuleRegistry::instance(); + if (action == QStringLiteral("add_filter")) { + softbus::message_bus::pipeline::FilterRule rule; + rule.id = params.value(QStringLiteral("id")).toString(); + rule.scope = parseScope(params); + rule.condition = params.value(QStringLiteral("condition")).toObject(); + if (!reg.addFilterRule(rule)) { + return {4001, QStringLiteral("invalid_filter_rule"), {}}; + } + 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); + return {ok ? 0 : 4041, ok ? QStringLiteral("ok") : QStringLiteral("filter_not_found"), + QJsonObject{{QStringLiteral("id"), id}}}; + } + if (action == QStringLiteral("add_transform_rule")) { + softbus::message_bus::pipeline::TransformRule rule; + rule.id = params.value(QStringLiteral("id")).toString(); + rule.scope = parseScope(params); + 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 (!reg.addTransformRule(rule)) { + return {4002, QStringLiteral("invalid_transform_rule"), {}}; + } + 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); + return {ok ? 0 : 4042, ok ? QStringLiteral("ok") : QStringLiteral("transform_rule_not_found"), + QJsonObject{{QStringLiteral("id"), id}}}; + } + if (action == QStringLiteral("list_rules") || action == QStringLiteral("get_status")) { + return {0, QStringLiteral("ok"), reg.status()}; + } + return {4040, QStringLiteral("unknown_action"), {}}; +} + +} // namespace softbus::api::ipc + diff --git a/src/api/ipc_dbus/CommandDispatcher.h b/src/api/ipc_dbus/CommandDispatcher.h new file mode 100644 index 0000000..d49889a --- /dev/null +++ b/src/api/ipc_dbus/CommandDispatcher.h @@ -0,0 +1,23 @@ +#pragma once + +#include +#include + +namespace softbus::api::ipc +{ + +struct DispatchResult +{ + int code{0}; + QString message; + QJsonObject data; +}; + +class CommandDispatcher +{ +public: + DispatchResult dispatch(const QString& action, const QJsonObject& params) const; +}; + +} // namespace softbus::api::ipc + diff --git a/src/api/ipc_dbus/IpcDBus.cpp b/src/api/ipc_dbus/IpcDBus.cpp index da1cc8b..d3425b1 100644 --- a/src/api/ipc_dbus/IpcDBus.cpp +++ b/src/api/ipc_dbus/IpcDBus.cpp @@ -4,13 +4,16 @@ #include #include #include +#include +#include "api/ipc_dbus/CommandDispatcher.h" #include "src/core/CoreService.h" IpcDBus::IpcDBus(CoreService* coreService, const QString& role, QObject* parent) : QObject(parent), m_coreService(coreService), - m_role(role) + m_role(role), + m_dispatcher(std::make_unique()) { } @@ -68,6 +71,33 @@ bool IpcDBus::reloadConfig() return false; } +QString IpcDBus::executeCommand(const QString& requestJson) +{ + auto makeResponse = [](const QString& id, int code, const QString& message, const QJsonObject& data) { + QJsonObject o; + o.insert(QStringLiteral("id"), id); + o.insert(QStringLiteral("code"), code); + o.insert(QStringLiteral("message"), message); + o.insert(QStringLiteral("data"), data); + return QString::fromUtf8(QJsonDocument(o).toJson(QJsonDocument::Compact)); + }; + + QJsonParseError err; + const QJsonDocument doc = QJsonDocument::fromJson(requestJson.toUtf8(), &err); + if (err.error != QJsonParseError::NoError || !doc.isObject()) { + return makeResponse(QString(), 4000, QStringLiteral("invalid_json"), {}); + } + const QJsonObject req = doc.object(); + const QString id = req.value(QStringLiteral("id")).toString(); + const QString action = req.value(QStringLiteral("action")).toString(); + const QJsonObject params = req.value(QStringLiteral("params")).toObject(); + if (action.isEmpty() || !m_dispatcher) { + return makeResponse(id, 4003, QStringLiteral("invalid_action"), {}); + } + const auto result = m_dispatcher->dispatch(action, params); + return makeResponse(id, result.code, result.message, result.data); +} + void IpcDBus::updateLoadInfo() { // 预留：后续可在此周期性采样 CPU/内存/队列长度等指标，并通过 getStatus() 输出 diff --git a/src/api/ipc_dbus/IpcDBus.h b/src/api/ipc_dbus/IpcDBus.h index 0945c80..2827100 100644 --- a/src/api/ipc_dbus/IpcDBus.h +++ b/src/api/ipc_dbus/IpcDBus.h @@ -8,9 +8,11 @@ #include +#include #include // 前向声明 class CoreService; +namespace softbus::api::ipc { class CommandDispatcher; } /** * @brief D-Bus 服务适配器 - 暴露核心服务接口 */ @@ -54,6 +56,7 @@ public slots: bool reloadConfig(); + QString executeCommand(const QString& requestJson); @@ -64,6 +67,7 @@ private slots: private: CoreService* m_coreService{nullptr}; // non-owning QString m_role; + std::unique_ptr m_dispatcher; }; #endif // IPC_DBUS_H diff --git a/src/core/CoreService.cpp b/src/core/CoreService.cpp index 1f63344..57b4a83 100644 --- a/src/core/CoreService.cpp +++ b/src/core/CoreService.cpp @@ -1,5 +1,6 @@ #include "CoreService.h" +#include #include #include #include @@ -43,16 +44,18 @@ bool CoreService::initialize() // 1. 解析配置路径（放在用户 home 下的相对固定目录：~/softbus/config） const QString configPath = QDir::homePath() + QStringLiteral("/softbus/config/daemon_config.json"); + const QString deviceTreePath = + QDir::homePath() + QStringLiteral("/softbus/config/device_tree.json"); auto& bus = softbus::device_bus::DeviceBus::instance(); bus.setConfigPath(configPath); + bus.setDeviceTreePath(deviceTreePath); QJsonObject root; const QFileInfo fi(configPath); if (!fi.exists()) { LOG_WARNING() << "CoreService: config not found, skip load:" << configPath; - root.insert(QStringLiteral("device_tree"), QJsonArray{}); - bus.markDeviceTreeDirty(); // 首次启动：退出时至少会落一个最小配置文件 + root = QJsonObject{}; } else { QFile f(configPath); if (!f.open(QIODevice::ReadOnly | QIODevice::Text)) { @@ -72,6 +75,69 @@ bool CoreService::initialize() root = doc.object(); } + const QFileInfo treeFi(deviceTreePath); + if (treeFi.exists()) { + QFile treeFile(deviceTreePath); + if (!treeFile.open(QIODevice::ReadOnly | QIODevice::Text)) { + LOG_ERROR() << "CoreService: failed to open" << deviceTreePath; + return false; + } + const QByteArray treeData = treeFile.readAll(); + treeFile.close(); + + QJsonParseError treePe; + const QJsonDocument treeDoc = QJsonDocument::fromJson(treeData, &treePe); + if (treePe.error != QJsonParseError::NoError || !treeDoc.isArray()) { + LOG_ERROR() << "CoreService: invalid JSON array in" << deviceTreePath << ":" << treePe.errorString(); + return false; + } + root.insert(QStringLiteral("device_tree"), treeDoc.array()); + } else { + if (!root.contains(QStringLiteral("device_tree")) || !root.value(QStringLiteral("device_tree")).isArray()) { + root.insert(QStringLiteral("device_tree"), QJsonArray{}); + } + bus.markDeviceTreeDirty(); // 首次独立设备树文件缺失：退出时落一份初始文件 + } + + auto resolvePluginPaths = [&](const QString& key) { + QJsonObject plugins = root.value(QStringLiteral("plugins")).toObject(); + QJsonArray arr = plugins.value(key).toArray(); + if (arr.isEmpty()) { + return; + } + + const QString configDir = fi.dir().absolutePath(); + const QString appDir = QCoreApplication::applicationDirPath(); + for (int i = 0; i < arr.size(); ++i) { + QJsonObject obj = arr.at(i).toObject(); + const QString rawPath = obj.value(QStringLiteral("path")).toString(); + if (rawPath.isEmpty() || QFileInfo(rawPath).isAbsolute()) { + continue; + } + // 先尝试绝对路径，再尝试相对路径 TODO 后续可以优化为只尝试相对路径 + const QStringList candidates = { + QDir(configDir).filePath(rawPath), + QDir(configDir).filePath(QStringLiteral("../") + rawPath), + QDir(appDir).filePath(rawPath), + QDir(appDir).filePath(QStringLiteral("../") + rawPath), + }; + + QString resolved = rawPath; + for (const auto& c : candidates) { + if (QFileInfo::exists(c)) { + resolved = QFileInfo(c).absoluteFilePath(); + break; + } + } + obj.insert(QStringLiteral("path"), resolved); + arr[i] = obj; + } + + plugins.insert(key, arr); + root.insert(QStringLiteral("plugins"), plugins); + }; + resolvePluginPaths(QStringLiteral("protocols")); + resolvePluginPaths(QStringLiteral("framers")); // 2. 初始化 device_bus（交给 DeviceBus facade 处理具体设备） if (!bus.initialize(root)) { diff --git a/src/core/models/DOMMessage.h b/src/core/models/DOMMessage.h index 5e5a789..f102219 100644 --- a/src/core/models/DOMMessage.h +++ b/src/core/models/DOMMessage.h @@ -1,9 +1,12 @@ #pragma once // 定义DOM消息 #include +#include #include #include +#include +#include #include "core/models/MetadataDef.h" #include "core/models/Types.h" @@ -18,11 +21,13 @@ class DOMMessage public: DOMMessage() = default; - QString messageId; - const MetadataDef* metaRef{nullptr}; - std::int64_t timestamp{0}; - DOMValue value{0.0}; - DataQuality quality{DataQuality::BAD}; + QString messageId; // 消息ID + const MetadataDef* metaRef{nullptr}; // 元数据引用 + std::int64_t timestamp{0}; // 时间戳 + DOMValue value{0.0}; // 值 + DataQuality quality{DataQuality::BAD}; // 质量 + QHash attributes; // 属性 + std::vector traceLog; // 跟踪日志 DataDomain domain() const { return metaRef ? metaRef->domain : DataDomain::UNKNOWN; } @@ -34,6 +39,11 @@ public: } return T{}; } + + void appendTrace(const QString& log) + { + traceLog.push_back(log); + } }; } // namespace softbus::core::models diff --git a/src/core/models/EnrichedMessage.h b/src/core/models/EnrichedMessage.h new file mode 100644 index 0000000..a0f1e1b --- /dev/null +++ b/src/core/models/EnrichedMessage.h @@ -0,0 +1,25 @@ +#pragma once + +#include +#include +#include + +#include "core/models/DOMMessage.h" + +namespace softbus::core::models +{ + +struct EnrichedMessage +{ + QString sourceMessageId; + QString logicalName; + QString unit; + DOMValue engineeringValue{0.0}; + DataQuality quality{DataQuality::BAD}; + bool alarm{false}; + QHash attributes; + std::vector traceLog; +}; + +} // namespace softbus::core::models + diff --git a/src/core/models/MetadataDef.h b/src/core/models/MetadataDef.h index 42ec20b..075a8fb 100644 --- a/src/core/models/MetadataDef.h +++ b/src/core/models/MetadataDef.h @@ -9,16 +9,16 @@ namespace softbus::core::models struct MetadataDef { - QString metadataId; - QString name; - QString description; - DataDomain domain{DataDomain::UNKNOWN}; - DataType semanticType{DataType::FLOAT64}; - QString unit; - double minValue{0.0}; - double maxValue{0.0}; - double deadband{0.0}; - bool isWriteable{false}; + QString metadataId; // 元数据ID 唯一标识 + QString name; // 名称 + QString description; // 描述 + DataDomain domain{DataDomain::UNKNOWN}; // 数据域 + DataType semanticType{DataType::FLOAT64}; // 语义类型 + QString unit; // 单位 + double minValue{0.0}; // 最小值 + double maxValue{0.0}; // 最大值 + double deadband{0.0}; // 死区 + bool isWriteable{false}; // 是否可写 }; } // namespace softbus::core::models diff --git a/src/core/plugin_system/IProtocolMapperPlugin.h b/src/core/plugin_system/IProtocolMapperPlugin.h new file mode 100644 index 0000000..4c417c3 --- /dev/null +++ b/src/core/plugin_system/IProtocolMapperPlugin.h @@ -0,0 +1,36 @@ +#pragma once + +#include +#include + +#include + +#include "core/models/DOMMessage.h" +#include "message_bus/pipeline/PipelineContext.h" +#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" + +namespace softbus::core::plugin_system +{ + +class IProtocolMapperSession +{ +public: + virtual ~IProtocolMapperSession() = default; + virtual std::vector map( + const softbus::message_bus::pipeline::PipelineContext& rawContext, + const softbus::message_bus::pipeline::protocol::ProtocolEnvelope& envelope) = 0; + virtual void reset() = 0; +}; + +class IProtocolMapperPlugin +{ +public: + virtual ~IProtocolMapperPlugin() = default; + + virtual QString pluginId() const = 0; + virtual bool supports(const QString& protocolHint) const = 0; + virtual std::unique_ptr createMapperSession() = 0; +}; + +} // namespace softbus::core::plugin_system + diff --git a/src/core/plugin_system/IProtocolPlugin.h b/src/core/plugin_system/IProtocolPlugin.h index 23bff7e..2107e36 100644 --- a/src/core/plugin_system/IProtocolPlugin.h +++ b/src/core/plugin_system/IProtocolPlugin.h @@ -4,6 +4,7 @@ #include +#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" #include "message_bus/pipeline/PipelineContext.h" namespace softbus::core::plugin_system @@ -13,7 +14,8 @@ class IProtocolSession { public: virtual ~IProtocolSession() = default; - virtual bool feed(softbus::message_bus::pipeline::PipelineContext& ctx) = 0; + virtual bool parse(const softbus::message_bus::pipeline::PipelineContext& ctx, + softbus::message_bus::pipeline::protocol::ProtocolEnvelope& out) = 0; virtual void reset() = 0; }; diff --git a/src/core/plugin_system/PluginHost.cpp b/src/core/plugin_system/PluginHost.cpp index 03ba54e..2b07f03 100644 --- a/src/core/plugin_system/PluginHost.cpp +++ b/src/core/plugin_system/PluginHost.cpp @@ -3,6 +3,7 @@ #include #include "core/plugin_system/IProtocolFramerPlugin.h" +#include "core/plugin_system/IProtocolMapperPlugin.h" #include "core/plugin_system/IProtocolPlugin.h" namespace softbus::core::plugin_system @@ -14,6 +15,7 @@ namespace using GetAbiVersionFn = int (*)(); using CreateProtocolPluginFn = IProtocolPlugin* (*)(); using CreateFramerPluginFn = IProtocolFramerPlugin* (*)(); +using CreateMapperPluginFn = IProtocolMapperPlugin* (*)(); } // namespace @@ -67,10 +69,20 @@ bool PluginHost::loadPlugin(const QString& path, QString* err) m_manager->registerProtocolFramerPlugin(p); } } + auto* createMapper = reinterpret_cast(dlsym(handle, "softbus_create_mapper_plugin")); + QString mapperPluginId; + if (createMapper) { + std::shared_ptr p(createMapper()); + if (p) { + mapperPluginId = p->pluginId(); + m_manager->registerProtocolMapperPlugin(p); + } + } LoadedPlugin lp; lp.path = path; lp.protocolPluginId = protocolPluginId; lp.framerPluginId = framerPluginId; + lp.mapperPluginId = mapperPluginId; lp.abiVersion = abi; lp.handle = handle; m_loaded.insert(path, lp); @@ -105,6 +117,9 @@ bool PluginHost::unloadPlugin(const QString& path, QString* err) if (!it.value().framerPluginId.isEmpty()) { m_manager->unregisterProtocolFramerPlugin(it.value().framerPluginId); } + if (!it.value().mapperPluginId.isEmpty()) { + m_manager->unregisterProtocolMapperPlugin(it.value().mapperPluginId); + } } if (it.value().handle) { dlclose(it.value().handle); diff --git a/src/core/plugin_system/PluginHost.h b/src/core/plugin_system/PluginHost.h index ea59a13..0b6e87a 100644 --- a/src/core/plugin_system/PluginHost.h +++ b/src/core/plugin_system/PluginHost.h @@ -18,6 +18,7 @@ public: QString path; QString protocolPluginId; QString framerPluginId; + QString mapperPluginId; int abiVersion{1}; void* handle{nullptr}; }; diff --git a/src/core/plugin_system/PluginManager.cpp b/src/core/plugin_system/PluginManager.cpp index 70732a0..52afaee 100644 --- a/src/core/plugin_system/PluginManager.cpp +++ b/src/core/plugin_system/PluginManager.cpp @@ -49,4 +49,42 @@ void PluginManager::unregisterProtocolFramerPlugin(const QString& pluginId) m_protocolFramerPlugins.remove(pluginId); } +void PluginManager::registerProtocolMapperPlugin(std::shared_ptr plugin) +{ + if (!plugin) { + return; + } + m_protocolMapperPlugins.insert(plugin->pluginId(), std::move(plugin)); +} + +std::shared_ptr PluginManager::selectProtocolMapperPlugin(const QString& protocolHint) const +{ + for (auto it = m_protocolMapperPlugins.constBegin(); it != m_protocolMapperPlugins.constEnd(); ++it) { + if (it.value() && it.value()->supports(protocolHint)) { + return it.value(); + } + } + return {}; +} + +void PluginManager::unregisterProtocolMapperPlugin(const QString& pluginId) +{ + m_protocolMapperPlugins.remove(pluginId); +} + +QStringList PluginManager::registeredProtocolPluginIds() const +{ + return m_protocolPlugins.keys(); +} + +QStringList PluginManager::registeredProtocolFramerPluginIds() const +{ + return m_protocolFramerPlugins.keys(); +} + +QStringList PluginManager::registeredProtocolMapperPluginIds() const +{ + return m_protocolMapperPlugins.keys(); +} + } // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/PluginManager.h b/src/core/plugin_system/PluginManager.h index 6cb5965..2f25c17 100644 --- a/src/core/plugin_system/PluginManager.h +++ b/src/core/plugin_system/PluginManager.h @@ -4,8 +4,10 @@ #include #include +#include #include "core/plugin_system/IProtocolFramerPlugin.h" +#include "core/plugin_system/IProtocolMapperPlugin.h" #include "core/plugin_system/IProtocolPlugin.h" namespace softbus::core::plugin_system @@ -22,10 +24,17 @@ public: void registerProtocolFramerPlugin(std::shared_ptr plugin); std::shared_ptr selectProtocolFramerPlugin(const QString& protocolHint) const; void unregisterProtocolFramerPlugin(const QString& pluginId); + void registerProtocolMapperPlugin(std::shared_ptr plugin); + std::shared_ptr selectProtocolMapperPlugin(const QString& protocolHint) const; + void unregisterProtocolMapperPlugin(const QString& pluginId); + QStringList registeredProtocolPluginIds() const; + QStringList registeredProtocolFramerPluginIds() const; + QStringList registeredProtocolMapperPluginIds() const; private: QHash> m_protocolPlugins; QHash> m_protocolFramerPlugins; + QHash> m_protocolMapperPlugins; }; } // namespace softbus::core::plugin_system diff --git a/src/device_bus/DeviceBus.cpp b/src/device_bus/DeviceBus.cpp index e92366b..cf726c6 100644 --- a/src/device_bus/DeviceBus.cpp +++ b/src/device_bus/DeviceBus.cpp @@ -20,6 +20,7 @@ public: std::unique_ptr registry; std::shared_ptr treeModel; QString configPath; + QString deviceTreePath; bool deviceTreeDirty{false}; // 设备树是否脏了 }; @@ -97,33 +98,33 @@ void DeviceBus::shutdown() { return; } - if (d->configPath.isEmpty() || !d->treeModel) { + if (d->deviceTreePath.isEmpty() || !d->treeModel) { return; } - const QFileInfo fi(d->configPath); + const QFileInfo fi(d->deviceTreePath); const QString dirPath = fi.dir().absolutePath(); - LOG_INFO() << "DeviceBus: saving config to" << dirPath; + LOG_INFO() << "DeviceBus: saving device_tree to" << d->deviceTreePath; if (!QDir().mkpath(dirPath)) { LOG_ERROR() << "DeviceBus: failed to create config dir" << dirPath; return; } - const QJsonObject root = d->treeModel->toDaemonConfigRoot(); - const QByteArray bytes = QJsonDocument(root).toJson(QJsonDocument::Indented); + const QJsonArray tree = d->treeModel->toDeviceTreeJson(); + const QByteArray bytes = QJsonDocument(tree).toJson(QJsonDocument::Indented); - QSaveFile out(d->configPath); + QSaveFile out(d->deviceTreePath); if (!out.open(QIODevice::WriteOnly | QIODevice::Text)) { - LOG_ERROR() << "DeviceBus: failed to open config for write:" << d->configPath; + LOG_ERROR() << "DeviceBus: failed to open device tree for write:" << d->deviceTreePath; return; } if (out.write(bytes) != bytes.size()) { - LOG_ERROR() << "DeviceBus: failed to write config:" << d->configPath; + LOG_ERROR() << "DeviceBus: failed to write device tree:" << d->deviceTreePath; out.cancelWriting(); return; } if (!out.commit()) { - LOG_ERROR() << "DeviceBus: failed to commit config:" << d->configPath; + LOG_ERROR() << "DeviceBus: failed to commit device tree:" << d->deviceTreePath; return; } @@ -140,6 +141,16 @@ QString DeviceBus::configPath() const { return d->configPath; } +void DeviceBus::setDeviceTreePath(const QString& path) { + if (!d) return; + d->deviceTreePath = path; +} + +QString DeviceBus::deviceTreePath() const { + if (!d) return {}; + return d->deviceTreePath; +} + std::shared_ptr DeviceBus::deviceTreeModel() const { if (!d) return {}; return d->treeModel; diff --git a/src/device_bus/DeviceBus.h b/src/device_bus/DeviceBus.h index 7aaf7e8..d27db60 100644 --- a/src/device_bus/DeviceBus.h +++ b/src/device_bus/DeviceBus.h @@ -17,6 +17,8 @@ public: void setConfigPath(const QString& path); QString configPath() const; + void setDeviceTreePath(const QString& path); + QString deviceTreePath() const; // 共享设备树模型（单一真源） std::shared_ptr deviceTreeModel() const; diff --git a/src/device_bus/manager/DeviceBusManager.cpp b/src/device_bus/manager/DeviceBusManager.cpp index 355a49f..b18d71e 100644 --- a/src/device_bus/manager/DeviceBusManager.cpp +++ b/src/device_bus/manager/DeviceBusManager.cpp @@ -11,7 +11,6 @@ #include "message_bus/ingress/DriverIngressAdapter.h" #include "message_bus/pipeline/PipelineEngine.h" #include "core/plugin_system/PluginManager.h" -#include "plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h" #include "device_bus/manager/SerialDeviceManager.h" #include "devices/DeviceTypes.h" #include "utils/logs/logging.h" @@ -61,19 +60,38 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig) if (!m_pluginHost) { m_pluginHost = std::make_unique(m_pluginManager); } - m_pluginManager->registerProtocolPlugin(softbus::core::plugin_system::makeModbusRtuProtocolPlugin()); const auto pluginsObj = rootConfig.value(QStringLiteral("plugins")).toObject(); - const auto framerArr = pluginsObj.value(QStringLiteral("framers")).toArray(); - for (const auto& v : framerArr) { - const auto obj = v.toObject(); - const QString path = obj.value(QStringLiteral("path")).toString(); - if (path.isEmpty()) { - continue; - } - QString err; - if (!m_pluginHost->loadPlugin(path, &err)) { - LOG_WARNING() << "DeviceBusManager: framer plugin load failed path=" << path << "err=" << err; + QSet loadedPluginPaths; + auto loadPlugins = [&](const QJsonArray& arr, const char* sourceTag) { + for (const auto& v : arr) { + const auto obj = v.toObject(); + const QString path = obj.value(QStringLiteral("path")).toString(); + if (path.isEmpty() || loadedPluginPaths.contains(path)) { + continue; + } + QString err; + if (!m_pluginHost->loadPlugin(path, &err)) { + LOG_WARNING() << "DeviceBusManager: plugin load failed source=" << sourceTag + << "path=" << path << "err=" << err; + continue; + } + loadedPluginPaths.insert(path); } + }; + // New generic entry for protocol plugins; each .so may export parser/framer/mapper factories. + loadPlugins(pluginsObj.value(QStringLiteral("protocols")).toArray(), "protocols"); + // Backward compatibility: legacy key still supported. + loadPlugins(pluginsObj.value(QStringLiteral("framers")).toArray(), "framers"); + if (m_pluginManager) { + const auto parserIds = m_pluginManager->registeredProtocolPluginIds(); + const auto framerIds = m_pluginManager->registeredProtocolFramerPluginIds(); + const auto mapperIds = m_pluginManager->registeredProtocolMapperPluginIds(); + LOG_INFO() << "DeviceBusManager: registered parser plugins(" << parserIds.size() + << "):" << parserIds.join(QStringLiteral(", ")); + LOG_INFO() << "DeviceBusManager: registered framer plugins(" << framerIds.size() + << "):" << framerIds.join(QStringLiteral(", ")); + LOG_INFO() << "DeviceBusManager: registered mapper plugins(" << mapperIds.size() + << "):" << mapperIds.join(QStringLiteral(", ")); } if (m_pipelineEngine) { m_pipelineEngine->setMemoryPool(m_memoryPool); diff --git a/src/device_bus/manager/SerialDeviceManager.cpp b/src/device_bus/manager/SerialDeviceManager.cpp index 922ae7b..90b999b 100644 --- a/src/device_bus/manager/SerialDeviceManager.cpp +++ b/src/device_bus/manager/SerialDeviceManager.cpp @@ -98,8 +98,8 @@ bool initializeSerialDeviceWithDeps(const QString& endpoint, const QJsonObject& merged.insert(QStringLiteral("stopBits"), stopBitsVal); merged.insert(QStringLiteral("flowControl"), flowStr); - LOG_INFO() << "DeviceBusManager: init serial endpoint=" << endpoint - << "params=" << QJsonDocument(merged).toJson(QJsonDocument::Compact); + // LOG_INFO() << "DeviceBusManager: init serial endpoint=" << endpoint + // << "params=" << QJsonDocument(merged).toJson(QJsonDocument::Compact); persistSerialParams(deps.treeModel, endpoint, merged, deps.dirtyCb); diff --git a/src/device_bus/monitor/DeviceMonitorService.cpp b/src/device_bus/monitor/DeviceMonitorService.cpp index 8258a56..109d144 100644 --- a/src/device_bus/monitor/DeviceMonitorService.cpp +++ b/src/device_bus/monitor/DeviceMonitorService.cpp @@ -115,18 +115,18 @@ void DeviceMonitorService::onEvent(const DeviceEvent& ev) } // 第一版：仅日志 - LOG_INFO() << "DeviceMonitor event" - << "type=" << toString(ev.type) - << "action=" << toString(ev.action) - << "subsystem=" << ev.subsystem - << "devnode=" << ev.devnode - << "ifname=" << ev.ifname - << "ifindex=" << ev.ifindex - << "linkUp=" << ev.linkUp - << "vid=" << ev.vid - << "pid=" << ev.pid - << "serial=" << ev.serial - << "path=" << ev.devicePath; + // LOG_INFO() << "DeviceMonitor event" + // << "type=" << toString(ev.type) + // << "action=" << toString(ev.action) + // << "subsystem=" << ev.subsystem + // << "devnode=" << ev.devnode + // << "ifname=" << ev.ifname + // << "ifindex=" << ev.ifindex + // << "linkUp=" << ev.linkUp + // << "vid=" << ev.vid + // << "pid=" << ev.pid + // << "serial=" << ev.serial + // << "path=" << ev.devicePath; } } // namespace softbus::device_bus::monitor diff --git a/src/device_bus/monitor/UdevDeviceMonitor.cpp b/src/device_bus/monitor/UdevDeviceMonitor.cpp index 3134ff9..ba3653f 100644 --- a/src/device_bus/monitor/UdevDeviceMonitor.cpp +++ b/src/device_bus/monitor/UdevDeviceMonitor.cpp @@ -197,11 +197,11 @@ bool UdevDeviceMonitor::shouldKeepEvent(const DeviceEvent& ev) const return false; } if (isConsoleLikeTty(ev.devnode)) { - LOG_INFO() << "UdevDeviceMonitor: console like tty" << ev.devnode; + // LOG_INFO() << "UdevDeviceMonitor: console like tty" << ev.devnode; return false; } if (!isWhitelistedSerialDevnode(ev.devnode)) { - LOG_INFO() << "UdevDeviceMonitor: not whitelisted serial devnode" << ev.devnode; + // LOG_INFO() << "UdevDeviceMonitor: not whitelisted serial devnode" << ev.devnode; return false; } } diff --git a/src/message_bus/pipeline/DynamicRuleRegistry.cpp b/src/message_bus/pipeline/DynamicRuleRegistry.cpp new file mode 100644 index 0000000..5fe72e7 --- /dev/null +++ b/src/message_bus/pipeline/DynamicRuleRegistry.cpp @@ -0,0 +1,95 @@ +#include "message_bus/pipeline/DynamicRuleRegistry.h" + +#include +#include + +namespace softbus::message_bus::pipeline +{ + +DynamicRuleRegistry& DynamicRuleRegistry::instance() +{ + static DynamicRuleRegistry registry; + return registry; +} + +bool DynamicRuleRegistry::addFilterRule(const FilterRule& rule) +{ + if (rule.id.isEmpty()) { + return false; + } + std::unique_lock lk(m_mtx); + m_filterRules.insert(rule.id, rule); + return true; +} + +bool DynamicRuleRegistry::removeFilterRule(const QString& id) +{ + std::unique_lock lk(m_mtx); + return m_filterRules.remove(id) > 0; +} + +bool DynamicRuleRegistry::addTransformRule(const TransformRule& rule) +{ + if (rule.id.isEmpty()) { + return false; + } + std::unique_lock lk(m_mtx); + m_transformRules.insert(rule.id, rule); + return true; +} + +bool DynamicRuleRegistry::removeTransformRule(const QString& id) +{ + std::unique_lock lk(m_mtx); + return m_transformRules.remove(id) > 0; +} + +std::vector DynamicRuleRegistry::snapshotFilterRules( + const QString& endpoint, int deviceId, const QString& stableKey) 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) { + if (matchScope(it.value().scope, endpoint, deviceId, stableKey)) { + out.push_back(it.value()); + } + } + return out; +} + +std::vector DynamicRuleRegistry::snapshotTransformRules( + const QString& endpoint, int deviceId, const QString& stableKey) 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) { + if (matchScope(it.value().scope, endpoint, deviceId, stableKey)) { + out.push_back(it.value()); + } + } + return out; +} + +QJsonObject DynamicRuleRegistry::status() const +{ + std::shared_lock lk(m_mtx); + QJsonObject o; + o.insert(QStringLiteral("filterRuleCount"), m_filterRules.size()); + o.insert(QStringLiteral("transformRuleCount"), m_transformRules.size()); + return o; +} + +bool DynamicRuleRegistry::matchScope( + const RuleScope& scope, const QString& endpoint, int deviceId, const QString& stableKey) const +{ + // hybrid scope: global + endpoint/device/stableKey selective match + const bool endpointOk = scope.endpoint.isEmpty() || scope.endpoint == endpoint; + const bool deviceOk = scope.deviceId < 0 || scope.deviceId == deviceId; + const bool stableKeyOk = scope.stableKey.isEmpty() || scope.stableKey == stableKey; + return endpointOk && deviceOk && stableKeyOk; +} + +} // namespace softbus::message_bus::pipeline + diff --git a/src/message_bus/pipeline/DynamicRuleRegistry.h b/src/message_bus/pipeline/DynamicRuleRegistry.h new file mode 100644 index 0000000..1bb6a2f --- /dev/null +++ b/src/message_bus/pipeline/DynamicRuleRegistry.h @@ -0,0 +1,64 @@ +#pragma once + +#include +#include + +#include +#include +#include + +namespace softbus::message_bus::pipeline +{ + +struct RuleScope +{ + QString endpoint; + int deviceId{-1}; + QString stableKey; +}; + +struct FilterRule +{ + QString id; + RuleScope scope; + QJsonObject condition; +}; + +struct TransformRule +{ + QString id; + RuleScope scope; + double scale{1.0}; + double offset{0.0}; + QString unit; +}; + +class DynamicRuleRegistry +{ +public: + static DynamicRuleRegistry& instance(); + + bool addFilterRule(const FilterRule& rule); + bool removeFilterRule(const QString& id); + bool addTransformRule(const TransformRule& rule); + bool removeTransformRule(const QString& id); + + std::vector snapshotFilterRules( + const QString& endpoint, int deviceId, const QString& stableKey) const; + std::vector snapshotTransformRules( + const QString& endpoint, int deviceId, const QString& stableKey) const; + + QJsonObject status() const; + +private: + DynamicRuleRegistry() = default; + bool matchScope(const RuleScope& scope, const QString& endpoint, int deviceId, const QString& stableKey) const; + +private: + mutable std::shared_mutex m_mtx; + QHash m_filterRules; + QHash m_transformRules; +}; + +} // namespace softbus::message_bus::pipeline + diff --git a/src/message_bus/pipeline/PipelineContext.h b/src/message_bus/pipeline/PipelineContext.h index 214baa6..1a09e44 100644 --- a/src/message_bus/pipeline/PipelineContext.h +++ b/src/message_bus/pipeline/PipelineContext.h @@ -5,14 +5,10 @@ #include #include -#include #include #include -#include -#include "core/models/DOMMessage.h" #include "message_bus/pipeline/PayloadRef.h" -#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" namespace softbus::message_bus::pipeline { @@ -46,13 +42,6 @@ struct PipelineContext BusDirection direction{BusDirection::Unknown}; std::uint64_t frameSeq{0}; - /// 强类型协议解析结果（推荐下游使用）；多协议扩展时只增不改 variant。 - softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol; - - /// 兼容旧逻辑与日志；新代码请优先使用 `protocol`，勿依赖此处 JSON 字段名。 - QJsonObject parsed; - QVector domMessages; - QString framerError; // 成帧/Pipeline 旁路诊断（可选） }; } // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index b9dbc4c..233b79a 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -5,9 +5,6 @@ #include #include #include - -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h" #include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h" #include "utils/logs/logging.h" @@ -91,9 +88,6 @@ void PipelineEngine::loadRuntimeConfigFromJson(const QJsonObject& obj) std::max(64, obj.value(QStringLiteral("framedQueueMax")).toInt(4096))); m_rtConfig.maxReorderDepth = static_cast( std::max(16, obj.value(QStringLiteral("maxReorderDepth")).toInt(256))); - m_rtConfig.modbusMaxAdu = static_cast( - std::max(8, obj.value(QStringLiteral("modbusRtuMaxFrameBytes")).toInt(256))); - m_rtConfig.modbusInterFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0); } void PipelineEngine::start() @@ -110,13 +104,6 @@ void PipelineEngine::start() m_passthroughFramer = std::make_shared(m_pool); - if (m_pluginManager) { - softbus::message_bus::pipeline::protocol::modbusrtu::ModbusRtuFramerPluginConfig cfg; - cfg.maxAdu = m_rtConfig.modbusMaxAdu; - cfg.interFrameTimeoutMs = m_rtConfig.modbusInterFrameTimeoutMs; - m_pluginManager->registerProtocolFramerPlugin( - softbus::message_bus::pipeline::protocol::modbusrtu::makeModbusRtuFramerPlugin(m_pool, cfg)); - } // 如果启用并行管道 则创建帧队列和解析线程 如果启用有序出口 则创建合并队列和合并线程 // 并行管道是用来处理多个端点的数据流的 @@ -183,6 +170,14 @@ void PipelineEngine::stop() std::lock_guard lk(m_framerSessionMutex); m_framerSessions.clear(); } + { + std::lock_guard lk(m_parserSessionMutex); + m_parserSessions.clear(); + } + { + std::lock_guard lk(m_mapperSessionMutex); + m_mapperSessions.clear(); + } } // 功能 ：清空 ingress 队列 // 参数 ：无 @@ -294,10 +289,7 @@ void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector lk(m_framerSessionMutex); auto it = m_framerSessions.find(key); if (it == m_framerSessions.end() || !it.value()) { - QJsonObject params; - params.insert(QStringLiteral("modbusRtuMaxFrameBytes"), static_cast(m_rtConfig.modbusMaxAdu)); - params.insert(QStringLiteral("modbusRtuInterFrameTimeoutMs"), m_rtConfig.modbusInterFrameTimeoutMs); - auto s = plugin->createFramerSession(chunkIn.endpoint, params); + auto s = plugin->createFramerSession(chunkIn.endpoint, QJsonObject{}); m_framerSessions.insert( key, s ? std::shared_ptr(std::move(s)) : nullptr); it = m_framerSessions.find(key); @@ -454,44 +446,19 @@ bool PipelineEngine::runStages234(PipelineContext& ctx) if (ctx.frameKind != FrameKind::CompleteFrame) { return false; } - std::ostringstream oss; - oss << std::hex << std::setfill('0'); - for (std::size_t i = 0; i < ctx.payload.block->size; ++i) { - oss << std::setw(2) << static_cast(ctx.payload.block->data[i]); - if (i + 1 < ctx.payload.block->size) { - oss << ' '; + auto messages = stage2Parser(ctx); + LOG_INFO() << "PipelineEngine: stage2Parser messages size=" << messages.size(); + if (messages.empty()) { + return false; + } + for (auto& msg : messages) { + if (!stage3Filter(msg)) { + continue; } - } - const QString hexPreview = QString::fromStdString(oss.str()); - LOG_INFO() << "the data is"<< hexPreview; - LOG_INFO() << "the protocol hint is"<< ctx.protocolHint; - const QString directionText = (ctx.direction == BusDirection::Upstream) - ? QStringLiteral("Upstream") - : (ctx.direction == BusDirection::Downstream) - ? QStringLiteral("Downstream") - : QStringLiteral("Unknown"); - LOG_INFO() << "the direction is" << directionText; - if (!stage2Parser(ctx)) { - return false; - } - QJsonObject parsed = ctx.parsed; - LOG_INFO() << "the parsed is"<< parsed; - - softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol = ctx.protocol; - if (const auto* modbusPdu = std::get_if(&protocol.pdu)) { - LOG_INFO() << "the protocol text is" << softbus::message_bus::pipeline::protocol::toText(*modbusPdu); - } else { - LOG_INFO() << "the protocol text is" << QStringLiteral("ProtocolEnvelopePdu::monostate"); - } - // @TODO 以后需要测试是否丢包和解析延迟 - // m_count++; - // LOG_INFO() << "the count of command is" << m_count; - - if (!stage3Filter(ctx)) { - return false; - } - if (!stage4Validator(ctx)) { - return false; + if (!stage4Validator(msg)) { + continue; + } + (void)applyTransformAndEnrich(ctx, msg); } return true; } @@ -500,47 +467,97 @@ bool PipelineEngine::runStages234(PipelineContext& ctx) // 参数 ：ctx 数据 // 返回值 ：是否成功 // 逻辑 ：如果有效载荷无效 则返回false 如果都成功 则返回true -bool PipelineEngine::stage4Validator(PipelineContext& ctx) +bool PipelineEngine::stage4Validator(const softbus::core::models::DOMMessage& message) const { - if (!ctx.payload.valid()) { - return false; - } - if (ctx.endpoint.isEmpty()) { - return false; - } - if (!ctx.timestamp.isValid()) { - ctx.timestamp = QDateTime::currentDateTimeUtc(); - } - if (ctx.frameKind != FrameKind::CompleteFrame) { - return false; - } - return true; + return !message.messageId.isEmpty(); } // 功能 ：运行阶段2解析器 // 参数 ：ctx 数据 // 返回值 ：是否成功 // 逻辑 ：如果插件管理器为空 则返回true 如果插件管理器不为空 则选择合适的插件 创建会话 如果会话为空 则返回false 如果会话不为空 则返回true -bool PipelineEngine::stage2Parser(PipelineContext& ctx) +std::vector PipelineEngine::stage2Parser(const PipelineContext& ctx) { if (!m_pluginManager) { - return true; + return {}; } - auto plugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint); - if (!plugin) { - return true; + auto parserPlugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint); + if (!parserPlugin) { + return {}; } - auto session = plugin->createSession(); - if (!session) { - return false; + const QString parserKey = ctx.endpoint + QStringLiteral("|") + parserPlugin->pluginId(); + std::shared_ptr parserSession; + { + std::lock_guard lk(m_parserSessionMutex); + auto it = m_parserSessions.find(parserKey); + if (it == m_parserSessions.end() || !it.value()) { + auto s = parserPlugin->createSession(); + m_parserSessions.insert( + parserKey, + s ? std::shared_ptr(std::move(s)) : nullptr); + it = m_parserSessions.find(parserKey); + } + if (it == m_parserSessions.end() || !it.value()) { + return {}; + } + parserSession = it.value(); } - return session->feed(ctx); + softbus::message_bus::pipeline::protocol::ProtocolEnvelope envelope; // 协议封装 + if (!parserSession->parse(ctx, envelope)) { + return {}; + } + + auto mapperPlugin = m_pluginManager->selectProtocolMapperPlugin(ctx.protocolHint); + if (!mapperPlugin) { + return {}; + } + const QString key = ctx.endpoint + QStringLiteral("|") + mapperPlugin->pluginId(); + std::shared_ptr session; + { + std::lock_guard lk(m_mapperSessionMutex); + auto it = m_mapperSessions.find(key); + if (it == m_mapperSessions.end() || !it.value()) { + auto s = mapperPlugin->createMapperSession(); + m_mapperSessions.insert( + key, + s ? std::shared_ptr(std::move(s)) : nullptr); + it = m_mapperSessions.find(key); + } + if (it == m_mapperSessions.end() || !it.value()) { + return {}; + } + session = it.value(); + } + return session->map(ctx, envelope); } -bool PipelineEngine::stage3Filter(PipelineContext& ctx) +bool PipelineEngine::stage3Filter(softbus::core::models::DOMMessage& message) { static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter; - return s_filter.apply(ctx); + return s_filter.process(message); +} + +softbus::core::models::EnrichedMessage PipelineEngine::applyTransformAndEnrich( + const PipelineContext& rawCtx, softbus::core::models::DOMMessage& message) const +{ + auto rules = softbus::message_bus::pipeline::DynamicRuleRegistry::instance().snapshotTransformRules( + rawCtx.endpoint, rawCtx.deviceId, rawCtx.stableKey); + for (const auto& rule : rules) { + const double base = std::visit([](const auto& x) { return static_cast(x); }, message.value); + const double next = base * rule.scale + rule.offset; + if (next != base) { + message.appendTrace(QStringLiteral("transform:%1 %2->%3") + .arg(rule.id) + .arg(base, 0, 'g', 12) + .arg(next, 0, 'g', 12)); + } + message.value = next; + if (!rule.unit.isEmpty()) { + message.attributes.insert(QStringLiteral("unit"), rule.unit); + } + } + static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter; + return s_filter.enrich(message); } } // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/PipelineEngine.h b/src/message_bus/pipeline/PipelineEngine.h index 36c8e5b..dcaf9cf 100644 --- a/src/message_bus/pipeline/PipelineEngine.h +++ b/src/message_bus/pipeline/PipelineEngine.h @@ -12,9 +12,13 @@ #include "core/plugin_system/PluginManager.h" #include "core/plugin_system/IProtocolFramerPlugin.h" +#include "core/plugin_system/IProtocolMapperPlugin.h" +#include "core/models/DOMMessage.h" +#include "core/models/EnrichedMessage.h" #include "core/memory/LockFreeQueue.h" #include "core/memory/MemoryPool.h" #include "core/threading/MutexQueue.h" +#include "message_bus/pipeline/DynamicRuleRegistry.h" #include "message_bus/pipeline/PipelineContext.h" #include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h" @@ -29,9 +33,6 @@ struct PipelineRuntimeConfig std::size_t framedQueueMax{4096}; std::size_t maxReorderDepth{256}; - - std::size_t modbusMaxAdu{256}; - int modbusInterFrameTimeoutMs{0}; }; struct EndpointFramingState; @@ -92,9 +93,11 @@ private: bool runStages234(PipelineContext& ctx); void ingestMergeItem(MergeItem item); - bool stage2Parser(PipelineContext& ctx); - bool stage3Filter(PipelineContext& ctx); - bool stage4Validator(PipelineContext& ctx); + std::vector stage2Parser(const PipelineContext& ctx); + bool stage3Filter(softbus::core::models::DOMMessage& message); + bool stage4Validator(const softbus::core::models::DOMMessage& message) const; + softbus::core::models::EnrichedMessage applyTransformAndEnrich( + const PipelineContext& rawCtx, softbus::core::models::DOMMessage& message) const; private: softbus::core::memory::LockFreeQueue m_ingressQ; @@ -110,6 +113,10 @@ private: std::shared_ptr m_passthroughFramer; std::mutex m_framerSessionMutex; QHash> m_framerSessions; + std::mutex m_parserSessionMutex; + QHash> m_parserSessions; + std::mutex m_mapperSessionMutex; + QHash> m_mapperSessions; std::mutex m_seqMutex; QHash m_nextFrameSeqByEndpoint; diff --git a/src/message_bus/pipeline/protocol/ProtocolEnvelope.h b/src/message_bus/pipeline/protocol/ProtocolEnvelope.h index bf47321..8c7ee4e 100644 --- a/src/message_bus/pipeline/protocol/ProtocolEnvelope.h +++ b/src/message_bus/pipeline/protocol/ProtocolEnvelope.h @@ -1,8 +1,9 @@ #pragma once +#include #include -#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" +#include "plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h" namespace softbus::message_bus::pipeline::protocol { @@ -11,10 +12,16 @@ enum class ProtocolFamily : std::uint8_t { None = 0, ModbusRtu, + CanOpen, }; -/// 流水线内协议解析结果的统一外壳；后续可在此 variant 中增加 CAN 等协议。 -using ProtocolEnvelopePdu = std::variant; +/// CANopen 协议占位类型（后续接入真实 CANopen 对象后替换/扩展）。 +struct CanOpenPduPlaceholder +{ +}; + +/// 流水线内协议解析结果的统一外壳；通过 variant 承载多协议强类型结果。 +using ProtocolEnvelopePdu = std::variant; struct ProtocolEnvelope { diff --git a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp index 329acfd..614d24b 100644 --- a/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp +++ b/src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp @@ -22,7 +22,6 @@ void PassthroughFramer::feed(softbus::message_bus::pipeline::PipelineContext& } auto blk = m_pool->allocate(n); if (!blk || !blk->data || blk->capacity < n) { - chunkIn.framerError = QStringLiteral("pool_exhausted"); return; } const std::uint8_t* src = chunkIn.payload.bytes(); diff --git a/src/message_bus/pipeline/stages/2_parsers/IParserStage.h b/src/message_bus/pipeline/stages/2_parsers/IParserStage.h index 52be947..b2ed7f2 100644 --- a/src/message_bus/pipeline/stages/2_parsers/IParserStage.h +++ b/src/message_bus/pipeline/stages/2_parsers/IParserStage.h @@ -1,5 +1,8 @@ #pragma once // 定义解析器阶段接口 +#include + +#include "core/models/DOMMessage.h" #include "message_bus/pipeline/PipelineContext.h" namespace softbus::message_bus::pipeline::stages::parsers @@ -9,7 +12,8 @@ class IParserStage { public: virtual ~IParserStage() = default; - virtual bool parse(softbus::message_bus::pipeline::PipelineContext& ctx) = 0; + virtual std::vector parse( + const softbus::message_bus::pipeline::PipelineContext& rawContext) = 0; }; } // namespace softbus::message_bus::pipeline::stages::parsers diff --git a/src/message_bus/pipeline/stages/3_filters/IFilterStage.h b/src/message_bus/pipeline/stages/3_filters/IFilterStage.h index 81741d4..2e3cddb 100644 --- a/src/message_bus/pipeline/stages/3_filters/IFilterStage.h +++ b/src/message_bus/pipeline/stages/3_filters/IFilterStage.h @@ -1,6 +1,6 @@ #pragma once -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/DOMMessage.h" namespace softbus::message_bus::pipeline::stages::filters { @@ -9,7 +9,7 @@ class IFilterStage { public: virtual ~IFilterStage() = default; - virtual bool apply(softbus::message_bus::pipeline::PipelineContext& ctx) = 0; + virtual bool process(softbus::core::models::DOMMessage& message) = 0; }; } // namespace softbus::message_bus::pipeline::stages::filters diff --git a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp index d8a65dc..1884fdf 100644 --- a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp +++ b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp @@ -1,565 +1,67 @@ #include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h" -#include -#include -#include -#include -#include -#include -#include - #include -#include #include -#include -#include - -#include - -#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" -#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" -#include "utils/logs/logging.h" namespace softbus::message_bus::pipeline::stages::filters { -namespace -{ - -using softbus::core::models::DOMMessage; -using softbus::core::models::DataPointMapping; -using softbus::core::models::DataQuality; -using softbus::core::models::DataType; -using softbus::core::models::Endianness; -using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; -using softbus::message_bus::pipeline::protocol::ProtocolFamily; -using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; - -QByteArray registersToBytes(const QVector& regs) -{ - QByteArray out; - out.reserve(regs.size() * 2); - for (std::uint16_t v : regs) { - out.append(static_cast((v >> 8) & 0xFF)); - out.append(static_cast(v & 0xFF)); - } - return out; -} - -QJsonArray registersToQJsonArray(const QVector& regs) -{ - QJsonArray a; - for (std::uint16_t v : regs) { - a.append(static_cast(v)); - } - return a; -} - -std::uint64_t readUnsignedRaw(const QByteArray& bytes, const DataPointMapping& mapping, bool* ok) -{ - *ok = false; - if (mapping.byteLength == 0 || mapping.byteLength > 8) { - return 0; - } - const int offset = static_cast(mapping.byteOffset); - const int len = static_cast(mapping.byteLength); - if (offset < 0 || offset + len > bytes.size()) { - return 0; - } - - std::array buf{}; - for (int i = 0; i < len; ++i) { - buf[static_cast(i)] = static_cast(bytes[offset + i]); - } - - if (mapping.endianness == Endianness::LittleEndian) { - std::reverse(buf.begin(), buf.begin() + len); - } else if (mapping.endianness == Endianness::BigEndianByteSwap && (len % 2 == 0)) { - for (int i = 0; i + 1 < len; i += 2) { - std::swap(buf[static_cast(i)], buf[static_cast(i + 1)]); - } - } - - std::uint64_t v = 0; - for (int i = 0; i < len; ++i) { - v = (v << 8) | buf[static_cast(i)]; - } - *ok = true; - return v; -} - -double decodeRawToDouble(std::uint64_t raw, const DataPointMapping& mapping) -{ - switch (mapping.rawType) { - case DataType::BOOL: - return raw ? 1.0 : 0.0; - case DataType::INT32: - return static_cast(static_cast(raw & 0xFFFFFFFFu)); - case DataType::UINT32: - return static_cast(static_cast(raw & 0xFFFFFFFFu)); - case DataType::INT64: - return static_cast(static_cast(raw)); - case DataType::FLOAT32: { - const std::uint32_t u = static_cast(raw & 0xFFFFFFFFu); - float f = 0.0f; - std::memcpy(&f, &u, sizeof(float)); - return static_cast(f); - } - case DataType::FLOAT64: { - double d = 0.0; - std::memcpy(&d, &raw, sizeof(double)); - return d; - } - case DataType::STRING: - case DataType::BYTE_ARRAY: - default: - return static_cast(raw); - } -} - -class ExprParser -{ -public: - ExprParser(const QString& expr, const QJsonArray& regs, double x) - : m_expr(expr) - , m_regs(regs) - , m_x(x) - { - } - - bool eval(double* out) - { - m_pos = 0; - if (!out) { - return false; - } - const auto v = parseBitOr(); - skipSpaces(); - if (!v.has_value() || m_pos != m_expr.size()) { - return false; - } - *out = *v; - return true; - } - -private: - std::optional parseBitOr() - { - auto lhs = parseBitAnd(); - if (!lhs.has_value()) { - return std::nullopt; - } - while (true) { - skipSpaces(); - if (!match('|')) { - break; - } - auto rhs = parseBitAnd(); - if (!rhs.has_value()) { - return std::nullopt; - } - const auto lv = static_cast(*lhs); - const auto rv = static_cast(*rhs); - lhs = static_cast(lv | rv); - } - return lhs; - } - - std::optional parseBitAnd() - { - auto lhs = parseShift(); - if (!lhs.has_value()) { - return std::nullopt; - } - while (true) { - skipSpaces(); - if (!match('&')) { - break; - } - auto rhs = parseShift(); - if (!rhs.has_value()) { - return std::nullopt; - } - const auto lv = static_cast(*lhs); - const auto rv = static_cast(*rhs); - lhs = static_cast(lv & rv); - } - return lhs; - } - - std::optional parseShift() - { - auto lhs = parseAddSub(); - if (!lhs.has_value()) { - return std::nullopt; - } - while (true) { - skipSpaces(); - if (match(QStringLiteral("<<"))) { - auto rhs = parseAddSub(); - if (!rhs.has_value()) { - return std::nullopt; - } - lhs = static_cast(static_cast(*lhs) << static_cast(*rhs)); - continue; - } - if (match(QStringLiteral(">>"))) { - auto rhs = parseAddSub(); - if (!rhs.has_value()) { - return std::nullopt; - } - lhs = static_cast(static_cast(*lhs) >> static_cast(*rhs)); - continue; - } - break; - } - return lhs; - } - - std::optional parseAddSub() - { - auto lhs = parseMulDiv(); - if (!lhs.has_value()) { - return std::nullopt; - } - while (true) { - skipSpaces(); - if (match('+')) { - auto rhs = parseMulDiv(); - if (!rhs.has_value()) { - return std::nullopt; - } - lhs = *lhs + *rhs; - continue; - } - if (match('-')) { - auto rhs = parseMulDiv(); - if (!rhs.has_value()) { - return std::nullopt; - } - lhs = *lhs - *rhs; - continue; - } - break; - } - return lhs; - } - - std::optional parseMulDiv() - { - auto lhs = parseUnary(); - if (!lhs.has_value()) { - return std::nullopt; - } - while (true) { - skipSpaces(); - if (match('*')) { - auto rhs = parseUnary(); - if (!rhs.has_value()) { - return std::nullopt; - } - lhs = *lhs * *rhs; - continue; - } - if (match('/')) { - auto rhs = parseUnary(); - if (!rhs.has_value() || *rhs == 0.0) { - return std::nullopt; - } - lhs = *lhs / *rhs; - continue; - } - break; - } - return lhs; - } - - std::optional parseUnary() - { - skipSpaces(); - if (match('-')) { - auto v = parseUnary(); - return v.has_value() ? std::optional(-*v) : std::nullopt; - } - if (match('+')) { - return parseUnary(); - } - return parsePrimary(); - } - - std::optional parsePrimary() - { - skipSpaces(); - if (match('(')) { - auto v = parseBitOr(); - skipSpaces(); - if (!v.has_value() || !match(')')) { - return std::nullopt; - } - return v; - } - - if (peekIsAlpha()) { - return parseVariable(); - } - return parseNumber(); - } - - std::optional parseVariable() - { - const auto name = parseIdentifier(); - if (name == QStringLiteral("x")) { - return m_x; - } - if (name.startsWith(QStringLiteral("r"))) { - bool ok = false; - const int idx = name.mid(1).toInt(&ok); - if (ok && idx >= 0 && idx < m_regs.size()) { - return static_cast(m_regs[idx].toInt(0)); - } - } - return std::nullopt; - } - - std::optional parseNumber() - { - skipSpaces(); - const int start = m_pos; - bool dot = false; - while (m_pos < m_expr.size()) { - const QChar c = m_expr[m_pos]; - if (c.isDigit()) { - ++m_pos; - continue; - } - if (c == QChar('.')) { - if (dot) { - break; - } - dot = true; - ++m_pos; - continue; - } - break; - } - if (m_pos <= start) { - return std::nullopt; - } - bool ok = false; - const double v = m_expr.mid(start, m_pos - start).toDouble(&ok); - if (!ok) { - return std::nullopt; - } - return v; - } - - QString parseIdentifier() - { - skipSpaces(); - const int start = m_pos; - while (m_pos < m_expr.size()) { - const QChar c = m_expr[m_pos]; - if (c.isLetterOrNumber() || c == QChar('_')) { - ++m_pos; - continue; - } - break; - } - return m_expr.mid(start, m_pos - start); - } - - bool match(QChar c) - { - if (m_pos < m_expr.size() && m_expr[m_pos] == c) { - ++m_pos; - return true; - } - return false; - } - - bool match(const QString& token) - { - if (m_expr.mid(m_pos, token.size()) == token) { - m_pos += token.size(); - return true; - } - return false; - } - - bool peekIsAlpha() const - { - return m_pos < m_expr.size() && (m_expr[m_pos].isLetter() || m_expr[m_pos] == QChar('_')); - } - - void skipSpaces() - { - while (m_pos < m_expr.size() && m_expr[m_pos].isSpace()) { - ++m_pos; - } - } - -private: - QString m_expr; - QJsonArray m_regs; - double m_x{0.0}; - int m_pos{0}; -}; - -softbus::core::models::DOMValue castToSemantic(double value, DataType t) -{ - using softbus::core::models::DOMValue; - switch (t) { - case DataType::BOOL: - return DOMValue{value != 0.0}; - case DataType::INT32: - return DOMValue{static_cast(value)}; - case DataType::UINT32: - return DOMValue{static_cast(value)}; - case DataType::INT64: - return DOMValue{static_cast(value)}; - case DataType::FLOAT32: - return DOMValue{static_cast(value)}; - case DataType::FLOAT64: - default: - return DOMValue{value}; - } -} - -DataQuality qualityByRange(double value, const softbus::core::models::MetadataDef& meta) -{ - if (!std::isfinite(value)) { - return DataQuality::BAD; - } - if (meta.maxValue > meta.minValue && (value < meta.minValue || value > meta.maxValue)) { - return DataQuality::UNCERTAIN; - } - return DataQuality::GOOD; -} - -QString normalizedProtocol(const QString& protocolHint) -{ - return protocolHint.trimmed().toLower(); -} - -} // namespace - ProtocolParseFilter::ProtocolParseFilter() = default; -bool ProtocolParseFilter::apply(softbus::message_bus::pipeline::PipelineContext& ctx) +bool ProtocolParseFilter::process(softbus::core::models::DOMMessage& message) { - if (ctx.direction == softbus::message_bus::pipeline::BusDirection::Downstream) { + if (!ensureLoaded()) { return true; } - if (!ensureLoaded()) { - return true; // 保持兼容：配置缺失不阻断流水线 + // Message-stage filter: only normalize quality by metadata range. + if (!message.metaRef) { + return true; } - if (normalizedProtocol(ctx.protocolHint).contains(QStringLiteral("modbus"))) { - return parseModbus(ctx); + const auto* meta = m_metadataRegistry.findById(message.metaRef->metadataId); + if (!meta) { + return true; + } + const double v = std::visit( + [](const auto& x) { return static_cast(x); }, + message.value); + if (meta->maxValue > meta->minValue && (v < meta->minValue || v > meta->maxValue)) { + message.quality = softbus::core::models::DataQuality::UNCERTAIN; + message.appendTrace(QStringLiteral("filter:range_check -> uncertain")); } return true; } -bool ProtocolParseFilter::ensureLoaded() +softbus::core::models::EnrichedMessage +ProtocolParseFilter::enrich(const softbus::core::models::DOMMessage& message) +{ + softbus::core::models::EnrichedMessage out; + out.sourceMessageId = message.messageId; + out.logicalName = message.metaRef ? message.metaRef->name : message.messageId; + out.unit = message.metaRef ? message.metaRef->unit : QString(); + out.engineeringValue = message.value; + out.quality = message.quality; + out.attributes = message.attributes; + out.traceLog = message.traceLog; + return out; +} + +bool ProtocolParseFilter::ensureLoaded() const { std::call_once(m_loadOnce, [this]() { QString configBase = qEnvironmentVariable("SOFTBUS_CONFIG_DIR"); if (!configBase.isEmpty() && !configBase.endsWith(QChar('/'))) { configBase.append(QChar('/')); } - const QString homeCfg = configBase.isEmpty() - ? (QDir::homePath() + QStringLiteral("/softbus/config/")) - : configBase; - const QString metaHome = homeCfg + QStringLiteral("metadata_dictionary.json"); - const QString profHome = homeCfg + QStringLiteral("device_profiles.json"); + const QString homeCfg = configBase.isEmpty() + ? (QDir::homePath() + QStringLiteral("/softbus/config/")) + : configBase; + const QString metaHome = homeCfg + QStringLiteral("metadata_dictionary.json"); const QString metaLocal = QStringLiteral("config/metadata_dictionary.json"); - const QString profLocal = QStringLiteral("config/device_profiles.json"); - const QString metaPath = QFileInfo::exists(metaHome) ? metaHome : metaLocal; - const QString profPath = QFileInfo::exists(profHome) ? profHome : profLocal; - - const bool metaOk = m_metadataRegistry.loadFromFile(metaPath); - const bool profOk = m_profileRegistry.loadFromFile(profPath); - if (!metaOk || !profOk) { - LOG_WARNING() << "ProtocolParseFilter: config load failed metadata=" - << m_metadataRegistry.lastError() << " profile=" - << m_profileRegistry.lastError(); - m_loaded = false; - return; - } - m_loaded = true; + m_loaded = m_metadataRegistry.loadFromFile(metaPath); }); return m_loaded; } -bool ProtocolParseFilter::parseModbus(softbus::message_bus::pipeline::PipelineContext& ctx) -{ - QVector regsVec; - QJsonArray regsForExpr; - - if (ctx.protocol.family == ProtocolFamily::ModbusRtu) { - if (const auto* m = std::get_if(&ctx.protocol.pdu)) { - if (const auto* resp = std::get_if(&*m)) { - regsVec = resp->registers; - regsForExpr = registersToQJsonArray(regsVec); - } - } - } - if (regsVec.isEmpty()) { - const QJsonArray regsLegacy = ctx.parsed.value(QStringLiteral("registers")).toArray(); - if (regsLegacy.isEmpty()) { - return true; - } - regsForExpr = regsLegacy; - for (const auto& reg : regsLegacy) { - regsVec.append(static_cast(reg.toInt(0) & 0xFFFF)); - } - } - - const auto* profile = m_profileRegistry.find(ctx.stableKey, ctx.protocolHint); - if (!profile) { - profile = m_profileRegistry.find(QString::number(ctx.deviceId), ctx.protocolHint); - } - if (!profile) { - return true; - } - - const QByteArray bytes = registersToBytes(regsVec); - ctx.domMessages.clear(); - ctx.domMessages.reserve(profile->mappings.size()); - - for (const auto& mapping : profile->mappings) { - const auto* meta = m_metadataRegistry.findById(mapping.metadataId); - if (!meta) { - continue; - } - - bool ok = false; - const std::uint64_t raw = readUnsignedRaw(bytes, mapping, &ok); - if (!ok) { - continue; - } - - double v = decodeRawToDouble(raw, mapping); - v = v * mapping.scaleFactor + mapping.offsetFactor; - if (!mapping.expression.trimmed().isEmpty()) { - ExprParser parser(mapping.expression, regsForExpr, v); - double exprV = 0.0; - if (parser.eval(&exprV)) { - v = exprV; - } - } - - DOMMessage msg; - msg.messageId = QStringLiteral("%1.%2") - .arg(ctx.stableKey.isEmpty() ? QString::number(ctx.deviceId) : ctx.stableKey, - mapping.pointId); - msg.metaRef = meta; - msg.timestamp = ctx.timestamp.toMSecsSinceEpoch(); - msg.value = castToSemantic(v, meta->semanticType); - msg.quality = qualityByRange(v, *meta); - ctx.domMessages.append(std::move(msg)); - } - return true; -} - } // namespace softbus::message_bus::pipeline::stages::filters + diff --git a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h index 7b15a89..d81b237 100644 --- a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h +++ b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h @@ -2,8 +2,8 @@ #include +#include "core/models/EnrichedMessage.h" #include "core/metadata/MetadataRegistry.h" -#include "core/metadata/ProfileRegistry.h" #include "message_bus/pipeline/stages/3_filters/IFilterStage.h" namespace softbus::message_bus::pipeline::stages::filters @@ -14,17 +14,16 @@ class ProtocolParseFilter final : public IFilterStage public: ProtocolParseFilter(); - bool apply(softbus::message_bus::pipeline::PipelineContext& ctx) override; + bool process(softbus::core::models::DOMMessage& message) override; + softbus::core::models::EnrichedMessage enrich(const softbus::core::models::DOMMessage& message); private: - bool ensureLoaded(); - bool parseModbus(softbus::message_bus::pipeline::PipelineContext& ctx); + bool ensureLoaded() const; private: - softbus::core::metadata::MetadataRegistry m_metadataRegistry; - softbus::core::metadata::ProfileRegistry m_profileRegistry; - std::once_flag m_loadOnce; - bool m_loaded{false}; + mutable softbus::core::metadata::MetadataRegistry m_metadataRegistry; + mutable std::once_flag m_loadOnce; + mutable bool m_loaded{false}; }; } // namespace softbus::message_bus::pipeline::stages::filters diff --git a/tests/modbus_rtu_framer_test.cpp b/tests/modbus_rtu_framer_test.cpp index 9b600a7..4e50200 100644 --- a/tests/modbus_rtu_framer_test.cpp +++ b/tests/modbus_rtu_framer_test.cpp @@ -1,5 +1,5 @@ #include "message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h" -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" #include diff --git a/tests/modbus_rtu_framing_predict_test.cpp b/tests/modbus_rtu_framing_predict_test.cpp index 3080f5d..85a231f 100644 --- a/tests/modbus_rtu_framing_predict_test.cpp +++ b/tests/modbus_rtu_framing_predict_test.cpp @@ -1,4 +1,4 @@ -#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" +#include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" #include diff --git a/tests/modbus_rtu_pdu_codec_test.cpp b/tests/modbus_rtu_pdu_codec_test.cpp index cf667a9..9ea6759 100644 --- a/tests/modbus_rtu_pdu_codec_test.cpp +++ b/tests/modbus_rtu_pdu_codec_test.cpp @@ -1,4 +1,4 @@ -#include "plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h" +#include "plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h" #include diff --git a/tests/protocol_parse_filter_test.cpp b/tests/protocol_parse_filter_test.cpp index 5431701..3cc995b 100644 --- a/tests/protocol_parse_filter_test.cpp +++ b/tests/protocol_parse_filter_test.cpp @@ -6,14 +6,7 @@ #include #include -#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" -#include "message_bus/pipeline/protocol/ProtocolEnvelope.h" - -using softbus::message_bus::pipeline::BusDirection; -using softbus::message_bus::pipeline::PipelineContext; -using softbus::message_bus::pipeline::protocol::ModbusRtuPdu; -using softbus::message_bus::pipeline::protocol::ProtocolFamily; -using softbus::message_bus::pipeline::protocol::ReadHoldingRegistersResponse; +#include "core/models/DOMMessage.h" using softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter; class ProtocolParseFilterTest : public QObject @@ -61,29 +54,14 @@ void ProtocolParseFilterTest::initTestCase() void ProtocolParseFilterTest::parses_multiple_dom_messages() { ProtocolParseFilter filter; - PipelineContext ctx; - ctx.direction = BusDirection::Upstream; - ctx.protocolHint = QStringLiteral("modbus_rtu"); - ctx.stableKey = QStringLiteral("devA"); - ctx.deviceId = 1; - ctx.timestamp = QDateTime::currentDateTimeUtc(); - - ReadHoldingRegistersResponse resp; - resp.unitId = 1; - resp.registers = QVector{0, 255, 0, 5000}; - ctx.protocol.family = ProtocolFamily::ModbusRtu; - ctx.protocol.pdu = ModbusRtuPdu{std::move(resp)}; - - QVERIFY(filter.apply(ctx)); - QCOMPARE(ctx.domMessages.size(), 2); - - QCOMPARE(ctx.domMessages[0].messageId, QStringLiteral("devA.temperature")); - QCOMPARE(ctx.domMessages[0].quality, softbus::core::models::DataQuality::GOOD); - QCOMPARE(ctx.domMessages[0].domain(), softbus::core::models::DataDomain::PROCESS_PARAM); - QCOMPARE(ctx.domMessages[0].getValueAs(), 25.5); - - QCOMPARE(ctx.domMessages[1].messageId, QStringLiteral("devA.pressure")); - QCOMPARE(ctx.domMessages[1].getValueAs(), 5.0); + softbus::core::models::DOMMessage msg; + msg.messageId = QStringLiteral("devA.temperature"); + msg.value = 25.5; + msg.quality = softbus::core::models::DataQuality::GOOD; + QVERIFY(filter.process(msg)); + auto enriched = filter.enrich(msg); + QCOMPARE(enriched.sourceMessageId, msg.messageId); + QCOMPARE(std::get(enriched.engineeringValue), 25.5); } QTEST_MAIN(ProtocolParseFilterTest)