diff --git a/CMakeLists.txt b/CMakeLists.txt index 05b3ed1..d0f9e8e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,254 +1,213 @@ -cmake_minimum_required(VERSION 3.19) -project(soft_bus_daemon - LANGUAGES CXX - VERSION 1.0.0 -) - -set(CMAKE_EXPORT_COMPILE_COMMANDS ON) - -set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g") -set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -g") - -# Output layout -set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin) -set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin/plugins) -set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin/plugins) - -# 启用AddressSanitizer（仅在非Windows平台上） -if(NOT WIN32) - set(SANITIZE_FLAGS "-fsanitize=address") - set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${SANITIZE_FLAGS}") - set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${SANITIZE_FLAGS}") - set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} ${SANITIZE_FLAGS}") -else() - message(STATUS "AddressSanitizer disabled on Windows because the runtime is unavailable in the MinGW kit.") -endif() - -# 设置C++标准 -set(CMAKE_CXX_STANDARD 20) -set(CMAKE_CXX_STANDARD_REQUIRED ON) -set(CMAKE_AUTORCC ON) -set(CMAKE_AUTOMOC ON) -# 查找Qt6（守护进程只需要 Core/DBus/Network/SerialPort/Sql） -find_package(Qt6 6.5 REQUIRED COMPONENTS Core SerialPort Network Sql DBus) -include(CheckIncludeFileCXX) - -# 检测编译平台 -if("${CMAKE_SIZEOF_VOID_P}" STREQUAL "4") - set(PlatformDir "x86") -else() - set(PlatformDir "x64") -endif() - -qt_standard_project_setup() - -# 添加软总线模板库 -# 1. 日志库 -qt_add_library(softbus_template - src/utils/logs/logging.h - src/utils/logs/LogStream.h -) - -qt_add_executable(softbus_daemon - main.cpp - src/core/CoreService.cpp - - - - - # api 接口 - - src/api/ipc_dbus/IpcDBus.h - src/api/ipc_dbus/IpcDBus.cpp - - # device types - src/devices/DeviceTypes.h - src/devices/DeviceTypes.cpp - - # hardware interfaces (Level 2) - src/hardware/interfaces/IStreamDriver.h - src/hardware/interfaces/IFrameDriver.h - src/hardware/interfaces/IDeviceWatcher.h - - # hardware drivers (Level 2) - src/hardware/drivers/serial/SerialQtDriver.h - src/hardware/drivers/serial/SerialQtDriver.cpp - src/hardware/drivers/serial/SerialCapabilities.h - - # device bus: tree & capabilities - src/device_bus/tree/DeviceTreeModel.h - src/device_bus/tree/DeviceTreeModel.cpp - - # device bus: monitor - src/device_bus/monitor/DeviceEvent.h - src/device_bus/monitor/IDeviceMonitor.h - src/device_bus/monitor/DeviceMonitorService.h - src/device_bus/monitor/DeviceMonitorService.cpp - src/device_bus/monitor/UdevDeviceMonitor.h - src/device_bus/monitor/UdevDeviceMonitor.cpp - src/device_bus/monitor/NetlinkDeviceMonitor.h - src/device_bus/monitor/NetlinkDeviceMonitor.cpp - src/device_bus/monitor/WindowsDeviceMonitor.h - src/device_bus/monitor/WindowsDeviceMonitor.cpp - - - # device bus: facade & manager - src/device_bus/DeviceBus.h - src/device_bus/DeviceBus.cpp - src/device_bus/manager/DeviceBusManager.h - src/device_bus/manager/DeviceBusManager.cpp - src/device_bus/manager/SerialDeviceManager.h - src/device_bus/manager/SerialDeviceManager.cpp - - # device bus: registry skeleton - src/device_bus/registry/IDeviceRegistry.h - src/device_bus/registry/DeviceRegistry.h - src/device_bus/registry/DeviceRegistry.cpp - - # devices: base skeleton - src/devices/base/IDevice.h - src/devices/base/IPhysicalDevice.h - - # devices: physical - src/devices/physical/SerialDevice.h - src/devices/physical/SerialDevice.cpp - - # core memory/plugin - src/core/memory/MemoryPool.h - src/core/memory/LockFreeQueue.h - src/core/plugin_system/IProtocolPlugin.h - src/core/plugin_system/PluginManager.h - src/core/plugin_system/PluginManager.cpp - src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h - src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp - src/core/models/Types.h - src/core/models/MetadataDef.h - src/core/models/DataMapping.h - src/core/models/DOMMessage.h - src/core/metadata/MetadataRegistry.h - src/core/metadata/MetadataRegistry.cpp - src/core/metadata/ProfileRegistry.h - src/core/metadata/ProfileRegistry.cpp - - # message bus pipeline - src/message_bus/pipeline/PayloadRef.h - src/message_bus/pipeline/PipelineContext.h - src/message_bus/pipeline/PipelineEngine.h - src/message_bus/pipeline/PipelineEngine.cpp - src/message_bus/pipeline/protocol/ProtocolEnvelope.h - src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h - src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h - src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp - src/message_bus/pipeline/stages/1_framers/IFramer.h - src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h - src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp - src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h - src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp - src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h - src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp - src/message_bus/pipeline/stages/2_parsers/IParserStage.h - src/message_bus/pipeline/stages/3_filters/IFilterStage.h - src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h - src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp - src/message_bus/pipeline/stages/4_validators/IValidator.h - - # message bus ingress/egress - src/message_bus/ingress/IIngressPort.h - src/message_bus/ingress/DriverIngressAdapter.h - src/message_bus/ingress/DriverIngressAdapter.cpp - src/message_bus/egress/IEgressPort.h - src/message_bus/egress/EgressRouter.h - src/message_bus/egress/EgressRouter.cpp -) - -target_include_directories(softbus_daemon - PRIVATE - ${CMAKE_SOURCE_DIR} - ${CMAKE_SOURCE_DIR}/src -) - - - -target_link_libraries(softbus_daemon - PRIVATE - Qt::Core - Qt6::DBus - Qt6::SerialPort - Qt6::Network -) - -# Platform-specific deps -if(UNIX AND NOT APPLE) - find_package(PkgConfig QUIET) - if(PkgConfig_FOUND) - pkg_check_modules(UDEV QUIET libudev) - endif() - - if(UDEV_FOUND) - target_include_directories(softbus_daemon PRIVATE ${UDEV_INCLUDE_DIRS}) - target_link_directories(softbus_daemon PRIVATE ${UDEV_LIBRARY_DIRS}) - target_link_libraries(softbus_daemon PRIVATE ${UDEV_LIBRARIES}) - else() - # Fallback: rely on system linker search path - target_link_libraries(softbus_daemon PRIVATE udev) - endif() -endif() - -if(WIN32) - target_link_libraries(softbus_daemon PRIVATE setupapi iphlpapi user32 advapi32) -endif() - -add_subdirectory(plugins) - -option(SOFTBUS_BUILD_TESTS "Build unit tests" OFF) -if(SOFTBUS_BUILD_TESTS) - find_package(Qt6 6.5 REQUIRED COMPONENTS Test) - qt_add_executable(softbus_modbus_rtu_framer_test - tests/modbus_rtu_framer_test.cpp - src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp - src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp - ) - target_include_directories(softbus_modbus_rtu_framer_test PRIVATE - ${CMAKE_SOURCE_DIR} - ${CMAKE_SOURCE_DIR}/src - ) - target_link_libraries(softbus_modbus_rtu_framer_test PRIVATE - Qt6::Core - Qt6::Test - ) - - qt_add_executable(softbus_protocol_parse_filter_test - tests/protocol_parse_filter_test.cpp - src/core/metadata/MetadataRegistry.cpp - src/core/metadata/ProfileRegistry.cpp - src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp - ) - - qt_add_executable(softbus_modbus_rtu_pdu_codec_test - tests/modbus_rtu_pdu_codec_test.cpp - src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp - src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp - ) - target_include_directories(softbus_modbus_rtu_pdu_codec_test PRIVATE - ${CMAKE_SOURCE_DIR} - ${CMAKE_SOURCE_DIR}/src - ) - target_link_libraries(softbus_modbus_rtu_pdu_codec_test PRIVATE - Qt6::Core - Qt6::Test - ) - target_include_directories(softbus_protocol_parse_filter_test PRIVATE - ${CMAKE_SOURCE_DIR} - ${CMAKE_SOURCE_DIR}/src - ) - target_link_libraries(softbus_protocol_parse_filter_test PRIVATE - Qt6::Core - Qt6::Test - ) -endif() - -install(TARGETS softbus_daemon - BUNDLE DESTINATION . - RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR} - LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} -) +cmake_minimum_required(VERSION 3.19) +project(soft_bus_daemon + LANGUAGES CXX + VERSION 1.0.0 +) + +set(CMAKE_EXPORT_COMPILE_COMMANDS ON) + +set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g") +set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -g") + +# Output layout +set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin) +set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin/plugins) +set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin/plugins) + +# 启用AddressSanitizer（仅在非Windows平台上） +if(NOT WIN32) + set(SANITIZE_FLAGS "-fsanitize=address") + set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${SANITIZE_FLAGS}") + set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${SANITIZE_FLAGS}") + set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} ${SANITIZE_FLAGS}") +else() + message(STATUS "AddressSanitizer disabled on Windows because the runtime is unavailable in the MinGW kit.") +endif() + +# 设置C++标准 +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_AUTORCC ON) +set(CMAKE_AUTOMOC ON) +# 查找Qt6（守护进程只需要 Core/DBus/Network/SerialPort/Sql） +find_package(Qt6 6.5 REQUIRED COMPONENTS Core SerialPort Network Sql DBus) +include(CheckIncludeFileCXX) + +# 检测编译平台 +if("${CMAKE_SIZEOF_VOID_P}" STREQUAL "4") + set(PlatformDir "x86") +else() + set(PlatformDir "x64") +endif() + +qt_standard_project_setup() + +# 添加软总线模板库 +# 1. 日志库 +qt_add_library(softbus_template + src/utils/logs/logging.h + src/utils/logs/LogStream.h +) + +qt_add_executable(softbus_daemon + main.cpp + src/core/CoreService.cpp + + + + + # api 接口 + + src/api/ipc_dbus/IpcDBus.h + src/api/ipc_dbus/IpcDBus.cpp + + # device types + src/devices/DeviceTypes.h + src/devices/DeviceTypes.cpp + + # hardware interfaces (Level 2) + src/hardware/interfaces/IStreamDriver.h + src/hardware/interfaces/IFrameDriver.h + src/hardware/interfaces/IDeviceWatcher.h + + # hardware drivers (Level 2) + src/hardware/drivers/serial/SerialQtDriver.h + src/hardware/drivers/serial/SerialQtDriver.cpp + src/hardware/drivers/serial/SerialCapabilities.h + + # device bus: tree & capabilities + src/device_bus/tree/DeviceTreeModel.h + src/device_bus/tree/DeviceTreeModel.cpp + + # device bus: monitor + src/device_bus/monitor/DeviceEvent.h + src/device_bus/monitor/IDeviceMonitor.h + src/device_bus/monitor/DeviceMonitorService.h + src/device_bus/monitor/DeviceMonitorService.cpp + src/device_bus/monitor/UdevDeviceMonitor.h + src/device_bus/monitor/UdevDeviceMonitor.cpp + src/device_bus/monitor/NetlinkDeviceMonitor.h + src/device_bus/monitor/NetlinkDeviceMonitor.cpp + src/device_bus/monitor/WindowsDeviceMonitor.h + src/device_bus/monitor/WindowsDeviceMonitor.cpp + + + # device bus: facade & manager + src/device_bus/DeviceBus.h + src/device_bus/DeviceBus.cpp + src/device_bus/manager/DeviceBusManager.h + src/device_bus/manager/DeviceBusManager.cpp + src/device_bus/manager/SerialDeviceManager.h + src/device_bus/manager/SerialDeviceManager.cpp + + # device bus: registry skeleton + src/device_bus/registry/IDeviceRegistry.h + src/device_bus/registry/DeviceRegistry.h + src/device_bus/registry/DeviceRegistry.cpp + + # devices: base skeleton + src/devices/base/IDevice.h + src/devices/base/IPhysicalDevice.h + + # devices: physical + src/devices/physical/SerialDevice.h + src/devices/physical/SerialDevice.cpp + + # core memory/plugin + src/core/memory/MemoryPool.h + src/core/memory/LockFreeQueue.h + src/core/plugin_system/IProtocolPlugin.h + src/core/plugin_system/IProtocolFramerPlugin.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 + src/core/models/Types.h + src/core/models/MetadataDef.h + src/core/models/DataMapping.h + src/core/models/DOMMessage.h + src/core/metadata/MetadataRegistry.h + src/core/metadata/MetadataRegistry.cpp + src/core/metadata/ProfileRegistry.h + src/core/metadata/ProfileRegistry.cpp + + # message bus pipeline + src/message_bus/pipeline/PayloadRef.h + src/message_bus/pipeline/PipelineContext.h + 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 + src/message_bus/pipeline/stages/1_framers/IFramer.h + + src/message_bus/pipeline/stages/1_framers/PassthroughFramer.h + src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp + src/message_bus/pipeline/stages/2_parsers/IParserStage.h + src/message_bus/pipeline/stages/3_filters/IFilterStage.h + src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h + src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp + src/message_bus/pipeline/stages/4_validators/IValidator.h + + # message bus ingress/egress + src/message_bus/ingress/IIngressPort.h + src/message_bus/ingress/DriverIngressAdapter.h + src/message_bus/ingress/DriverIngressAdapter.cpp + src/message_bus/egress/IEgressPort.h + src/message_bus/egress/EgressRouter.h + src/message_bus/egress/EgressRouter.cpp +) + +target_include_directories(softbus_daemon + PRIVATE + ${CMAKE_SOURCE_DIR} + ${CMAKE_SOURCE_DIR}/src +) + + + +target_link_libraries(softbus_daemon + PRIVATE + Qt::Core + Qt6::DBus + Qt6::SerialPort + Qt6::Network +) + +# Platform-specific deps +if(UNIX AND NOT APPLE) + find_package(PkgConfig QUIET) + if(PkgConfig_FOUND) + pkg_check_modules(UDEV QUIET libudev) + endif() + + if(UDEV_FOUND) + target_include_directories(softbus_daemon PRIVATE ${UDEV_INCLUDE_DIRS}) + target_link_directories(softbus_daemon PRIVATE ${UDEV_LIBRARY_DIRS}) + target_link_libraries(softbus_daemon PRIVATE ${UDEV_LIBRARIES}) + else() + # Fallback: rely on system linker search path + target_link_libraries(softbus_daemon PRIVATE udev) + endif() + target_link_libraries(softbus_daemon PRIVATE dl) +endif() + +if(WIN32) + target_link_libraries(softbus_daemon PRIVATE setupapi iphlpapi user32 advapi32) +endif() + +add_subdirectory(plugins) + + +install(TARGETS softbus_daemon + BUNDLE DESTINATION . + RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR} + LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} +) diff --git a/bin/softbus_daemon b/bin/softbus_daemon index 1f53231..3a3d729 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 index 737a653..f4a3446 100644 Binary files a/bin/softbus_modbus_rtu_framer_test and b/bin/softbus_modbus_rtu_framer_test differ diff --git a/bin/softbus_modbus_rtu_framing_predict_test b/bin/softbus_modbus_rtu_framing_predict_test new file mode 100644 index 0000000..c758ea8 Binary files /dev/null and b/bin/softbus_modbus_rtu_framing_predict_test differ diff --git a/bin/softbus_modbus_rtu_pdu_codec_test b/bin/softbus_modbus_rtu_pdu_codec_test index 32a4021..6f144b6 100644 Binary files a/bin/softbus_modbus_rtu_pdu_codec_test and b/bin/softbus_modbus_rtu_pdu_codec_test differ diff --git a/bin/softbus_protocol_parse_filter_test b/bin/softbus_protocol_parse_filter_test index 8f4eb79..d8c1409 100644 Binary files a/bin/softbus_protocol_parse_filter_test and b/bin/softbus_protocol_parse_filter_test differ diff --git a/config/daemon_config.json b/config/daemon_config.json index 985e3af..fdb63de 100644 --- a/config/daemon_config.json +++ b/config/daemon_config.json @@ -9,6 +9,19 @@ "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", diff --git a/plugins/protocols/modbus_rtu/CMakeLists.txt b/plugins/protocols/modbus_rtu/CMakeLists.txt index d7384e5..e97ef21 100644 --- a/plugins/protocols/modbus_rtu/CMakeLists.txt +++ b/plugins/protocols/modbus_rtu/CMakeLists.txt @@ -1,3 +1,14 @@ -# Placeholder for modbus_rtu plugin -add_library(modbus_rtu SHARED plugin_stub.cpp) -set_target_properties(modbus_rtu PROPERTIES OUTPUT_NAME "modbus_rtu") +# 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 +) +target_include_directories(modbus_rtu PRIVATE + ${CMAKE_SOURCE_DIR} + ${CMAKE_SOURCE_DIR}/src +) +target_link_libraries(modbus_rtu PRIVATE Qt6::Core) +set_target_properties(modbus_rtu PROPERTIES OUTPUT_NAME "modbus_rtu") diff --git a/plugins/protocols/modbus_rtu/plugin_stub.cpp b/plugins/protocols/modbus_rtu/plugin_stub.cpp index 1990c25..30f3acb 100644 --- a/plugins/protocols/modbus_rtu/plugin_stub.cpp +++ b/plugins/protocols/modbus_rtu/plugin_stub.cpp @@ -1,2 +1,17 @@ -extern "C" void -softbus_modbus_rtu_plugin_stub() {} +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h" +#include "plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h" + +extern "C" int softbus_plugin_abi_version() +{ + return 1; +} + +extern "C" softbus::core::plugin_system::IProtocolPlugin* softbus_create_protocol_plugin() +{ + return softbus::core::plugin_system::createModbusRtuProtocolPluginRaw(); +} + +extern "C" softbus::core::plugin_system::IProtocolFramerPlugin* softbus_create_framer_plugin() +{ + return softbus::message_bus::pipeline::protocol::modbusrtu::createModbusRtuFramerPluginRaw(); +} diff --git a/plugins/protocols/modbusrtu/README.md b/plugins/protocols/modbusrtu/README.md new file mode 100644 index 0000000..262ce7b --- /dev/null +++ b/plugins/protocols/modbusrtu/README.md @@ -0,0 +1,37 @@ +# Modbus RTU 协议模块（`modbusrtu`） + +本目录实现 **Modbus RTU** 的成帧、编解码与解析插件，供守护进程静态编译与 `plugins/protocols/modbus_rtu` 动态库（`.so`）共用同一套源码。 + +## 目录结构 + +| 子目录 | 职责 | +|--------|------| +| `types/` | PDU 数据模型：`ModbusRtuPdu` 及各类请求/响应结构体，供编解码与 `ProtocolEnvelope` 使用。 | +| `codec/` | ADU 与 `ModbusRtuPdu` 的互转：`decodeModbusRtu` / `encodeModbusRtu`（依赖 `types` 与 `framer` 中的 CRC 等工具函数）。 | +| `framer/` | 字节流成帧：`ModbusRtuFraming`（长度预测、CRC）、`ModbusRtuFramerPlugin`（Stage1 插件会话）。 | +| `parser/` | Stage2 协议插件：`ModbusRtuProtocolPlugin`，将完整帧解析为 `ProtocolEnvelope` / 兼容 JSON。 | + +## 依赖方向（建议遵守） + +``` +types （无本模块内依赖） + ↑ +codec、framer （可依赖 types；codec 使用 framer 中的 crc16/crcOk） + ↑ +parser （依赖 codec、types；不反向依赖 framer 插件类） +``` + +成帧与解析在流水线中分属不同阶段；**解析侧只应处理 `CompleteFrame`**，与 `PipelineContext::frameKind` 约定一致。 + +## 与动态库的关系 + +- 动态库入口：`plugins/protocols/modbus_rtu/plugin_stub.cpp`，导出 `softbus_create_protocol_plugin` / `softbus_create_framer_plugin` 等符号。 +- 主工程 `CMakeLists.txt` 同样列出本目录下源文件，用于无 `.so` 时的内置链接；路径与 include 均以 `plugins/protocols/modbusrtu/...` 为准。 + +## 配置提示 + +在 `daemon_config.json` 的 `plugins.framers` 中配置 `libmodbus_rtu.so` 路径即可加载成帧插件；解析插件加载方式与 `PluginHost` / 工程内注册逻辑一致。 + +## 扩展其它协议时的参考 + +新增协议（如 Modbus TCP、CANopen）时，建议沿用类似分层：**types → codec/framer → parser**，并在各自 `plugin_stub` 中导出统一 ABI 工厂函数，由 `PluginHost` 注册到 `PluginManager`。 diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp b/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.cpp similarity index 96% rename from src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp rename to plugins/protocols/modbusrtu/codec/ModbusRtuCodec.cpp index 807f520..313eb56 100644 --- a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp +++ b/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.cpp @@ -1,9 +1,9 @@ -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h" +#include "plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h" #include #include "message_bus/pipeline/PipelineContext.h" -#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" namespace softbus::message_bus::pipeline::protocol { @@ -12,12 +12,12 @@ namespace { using softbus::message_bus::pipeline::BusDirection; -using softbus::message_bus::pipeline::stages::framers::modbusRtuCrc16; -using softbus::message_bus::pipeline::stages::framers::modbusRtuCrcOk; +using softbus::message_bus::pipeline::protocol::modbusrtu::crc16; +using softbus::message_bus::pipeline::protocol::modbusrtu::crcOk; void appendCrcLe(QByteArray& adu) { - const auto crc = modbusRtuCrc16(reinterpret_cast(adu.constData()), + const auto crc = crc16(reinterpret_cast(adu.constData()), static_cast(adu.size())); adu.append(static_cast(crc & 0xFF)); adu.append(static_cast((crc >> 8) & 0xFF)); @@ -47,7 +47,7 @@ bool decodeModbusRtu(BusDirection dir, if (n < 5) { return fail("exception_incomplete"); } - if (!modbusRtuCrcOk(p, 5)) { + if (!crcOk(p, 5)) { return fail("exception_crc"); } out = ModbusExceptionPdu{p[0], p[1], p[2]}; @@ -57,7 +57,7 @@ bool decodeModbusRtu(BusDirection dir, const std::uint8_t unit = p[0]; const std::uint8_t fc = p[1]; auto parseReadReq = [&]() -> bool { - if (n != 8 || !modbusRtuCrcOk(p, 8)) { + if (n != 8 || !crcOk(p, 8)) { return false; } const std::uint16_t addr = static_cast((p[2] << 8) | p[3]); @@ -81,7 +81,7 @@ bool decodeModbusRtu(BusDirection dir, }; auto parseWriteSingleReqOrResp = [&]() -> bool { - if (n != 8 || !modbusRtuCrcOk(p, 8)) { + if (n != 8 || !crcOk(p, 8)) { return false; } const std::uint16_t addr = static_cast((p[2] << 8) | p[3]); @@ -116,7 +116,7 @@ bool decodeModbusRtu(BusDirection dir, if (n != static_cast(9) + byteCount) { return false; } - if (!modbusRtuCrcOk(p, n)) { + if (!crcOk(p, n)) { return false; } if (fc == 0x0F) { @@ -141,7 +141,7 @@ bool decodeModbusRtu(BusDirection dir, }; auto parseWriteMultiResp = [&]() -> bool { - if (n != 8 || !modbusRtuCrcOk(p, 8)) { + if (n != 8 || !crcOk(p, 8)) { return false; } const std::uint16_t start = static_cast((p[2] << 8) | p[3]); @@ -165,7 +165,7 @@ bool decodeModbusRtu(BusDirection dir, } const std::uint8_t bc = p[2]; const std::size_t total = static_cast(5) + bc; - if (n < total || !modbusRtuCrcOk(p, total)) { + if (n < total || !crcOk(p, total)) { return false; } if (fc == 0x01) { diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h b/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h similarity index 91% rename from src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h rename to plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h index 977db10..6534557 100644 --- a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h +++ b/plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h @@ -6,7 +6,7 @@ #include #include -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" +#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" namespace softbus::message_bus::pipeline { diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.cpp b/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.cpp new file mode 100644 index 0000000..f639d05 --- /dev/null +++ b/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.cpp @@ -0,0 +1,173 @@ +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h" + +#include +#include + +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" +#include "utils/logs/logging.h" + +namespace softbus::message_bus::pipeline::protocol::modbusrtu +{ + +namespace +{ + +class ModbusRtuFramerSession final : public softbus::core::plugin_system::IProtocolFramerSession +{ +public: + ModbusRtuFramerSession(std::shared_ptr pool, + ModbusRtuFramerPluginConfig cfg, + QString endpoint) + : m_pool(std::move(pool)) + , m_cfg(cfg) + , m_endpoint(std::move(endpoint)) + { + m_buf.resize(m_cfg.maxAdu); + } + + void feedChunk(softbus::message_bus::pipeline::PipelineContext& in, + std::vector& out) override + { + using softbus::message_bus::pipeline::FrameKind; + if (!m_pool || !in.payload.valid()) { + return; + } + const std::uint8_t* p = in.payload.bytes(); + const std::size_t n = in.payloadSize ? in.payloadSize : in.payload.length; + if (!p || n == 0) { + return; + } + m_lastRx = std::chrono::steady_clock::now(); + + if (m_cfg.interFrameTimeoutMs > 0 && m_len > 0) { + const auto age = std::chrono::duration_cast(m_lastRx - m_lastActivity).count(); + if (age >= m_cfg.interFrameTimeoutMs) { + m_len = 0; + } + } + m_lastActivity = m_lastRx; + + if (m_len + n > m_cfg.maxAdu) { + LOG_WARNING() << "ModbusRtuFramerPlugin: overflow endpoint=" << m_endpoint; + m_len = 0; + return; + } + std::memcpy(m_buf.data() + m_len, p, n); + m_len += n; + + while (m_len > 0) { + auto pred = predictAduLength(m_buf.data(), m_len, m_cfg.maxAdu); + if (!pred || m_len < *pred) { + break; + } + if (!crcOk(m_buf.data(), *pred)) { + if (m_len <= 1) { + m_len = 0; + break; + } + std::memmove(m_buf.data(), m_buf.data() + 1, m_len - 1); + --m_len; + continue; + } + + 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); + blk->size = *pred; + + softbus::message_bus::pipeline::PipelineContext framed; + framed.deviceId = in.deviceId; + framed.stableKey = in.stableKey; + framed.endpoint = in.endpoint; + framed.timestamp = in.timestamp; + framed.protocolHint = in.protocolHint; + framed.traceId = in.traceId; + framed.direction = in.direction; + framed.frameKind = FrameKind::CompleteFrame; + framed.payload.block = std::move(blk); + framed.payload.offset = 0; + framed.payload.length = *pred; + framed.payloadSize = *pred; + out.push_back(std::move(framed)); + + const std::size_t rest = m_len - *pred; + if (rest > 0) { + std::memmove(m_buf.data(), m_buf.data() + *pred, rest); + } + m_len = rest; + } + + in.payload = {}; + in.payloadSize = 0; + in.frameKind = FrameKind::StreamChunk; + } + + void reset() override { m_len = 0; } + void shutdown() override { m_len = 0; } + +private: + std::shared_ptr m_pool; + ModbusRtuFramerPluginConfig m_cfg; + QString m_endpoint; + std::vector m_buf; + std::size_t m_len{0}; + std::chrono::steady_clock::time_point m_lastRx{std::chrono::steady_clock::now()}; + std::chrono::steady_clock::time_point m_lastActivity{std::chrono::steady_clock::now()}; +}; + +} // namespace + +ModbusRtuFramerPlugin::ModbusRtuFramerPlugin(std::shared_ptr pool, + ModbusRtuFramerPluginConfig cfg) + : m_pool(std::move(pool)) + , m_cfg(cfg) +{ +} + +QString ModbusRtuFramerPlugin::pluginId() const +{ + return QStringLiteral("modbus_rtu_framer"); +} + +bool ModbusRtuFramerPlugin::supports(const QString& protocolHint) const +{ + return isModbusRtuHint(protocolHint); +} + +void ModbusRtuFramerPlugin::bindMemoryPool(std::shared_ptr pool) +{ + m_pool = std::move(pool); +} + +std::unique_ptr +ModbusRtuFramerPlugin::createFramerSession(const QString& endpoint, const QJsonObject& params) +{ + if (!m_pool) { + return {}; + } + ModbusRtuFramerPluginConfig cfg = m_cfg; + if (params.contains(QStringLiteral("modbusRtuMaxFrameBytes"))) { + cfg.maxAdu = static_cast( + std::max(8, params.value(QStringLiteral("modbusRtuMaxFrameBytes")).toInt(static_cast(cfg.maxAdu)))); + } + if (params.contains(QStringLiteral("modbusRtuInterFrameTimeoutMs"))) { + cfg.interFrameTimeoutMs = params.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(cfg.interFrameTimeoutMs); + } + return std::make_unique(m_pool, cfg, endpoint); +} + +std::shared_ptr makeModbusRtuFramerPlugin( + std::shared_ptr pool, ModbusRtuFramerPluginConfig cfg) +{ + return std::make_shared(std::move(pool), cfg); +} + +softbus::core::plugin_system::IProtocolFramerPlugin* createModbusRtuFramerPluginRaw() +{ + return new ModbusRtuFramerPlugin(nullptr, {}); +} + +} // namespace softbus::message_bus::pipeline::protocol::modbusrtu diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h b/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h new file mode 100644 index 0000000..45b257b --- /dev/null +++ b/plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h @@ -0,0 +1,44 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include + +#include "core/memory/MemoryPool.h" +#include "core/plugin_system/IProtocolFramerPlugin.h" + +namespace softbus::message_bus::pipeline::protocol::modbusrtu +{ + +struct ModbusRtuFramerPluginConfig +{ + std::size_t maxAdu{256}; + int interFrameTimeoutMs{0}; +}; + +class ModbusRtuFramerPlugin final : public softbus::core::plugin_system::IProtocolFramerPlugin +{ +public: + explicit ModbusRtuFramerPlugin(std::shared_ptr pool, + ModbusRtuFramerPluginConfig cfg = {}); + + QString pluginId() const override; + bool supports(const QString& protocolHint) const override; + void bindMemoryPool(std::shared_ptr pool) override; + std::unique_ptr createFramerSession( + const QString& endpoint, const QJsonObject& params) override; + +private: + std::shared_ptr m_pool; + ModbusRtuFramerPluginConfig m_cfg; +}; + +std::shared_ptr makeModbusRtuFramerPlugin( + std::shared_ptr pool, ModbusRtuFramerPluginConfig cfg = {}); +softbus::core::plugin_system::IProtocolFramerPlugin* createModbusRtuFramerPluginRaw(); + +} // namespace softbus::message_bus::pipeline::protocol::modbusrtu diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.cpp b/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.cpp new file mode 100644 index 0000000..b827d02 --- /dev/null +++ b/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.cpp @@ -0,0 +1,88 @@ +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" + +namespace softbus::message_bus::pipeline::protocol::modbusrtu +{ + +std::uint16_t crc16(const std::uint8_t* data, std::size_t len) +{ + std::uint16_t crc = 0xFFFF; + for (std::size_t i = 0; i < len; ++i) { + crc ^= data[i]; + for (int k = 0; k < 8; ++k) { + crc = (crc & 1) ? static_cast((crc >> 1) ^ 0xA001) + : static_cast(crc >> 1); + } + } + return crc; +} + +bool crcOk(const std::uint8_t* adu, std::size_t aduLen) +{ + if (!adu || aduLen < 4) { + return false; + } + const std::size_t body = aduLen - 2; + const std::uint16_t got = + static_cast(adu[body] | (static_cast(adu[body + 1]) << 8)); + return got == crc16(adu, body); +} + +static bool isReadFc(std::uint8_t fc) +{ + return fc == 0x01 || fc == 0x02 || fc == 0x03 || fc == 0x04; +} + +std::optional predictAduLength(const std::uint8_t* buf, std::size_t len, std::size_t maxAdu) +{ + if (!buf || len < 2) { + return std::nullopt; + } + const std::uint8_t fc = buf[1]; + if (fc & 0x80) { + return len >= 5 ? std::optional(5) : std::nullopt; + } + + if (isReadFc(fc)) { + if (len >= 8 && crcOk(buf, 8)) { + return std::optional(8); // read request + } + if (len < 3) { + return std::nullopt; + } + const std::size_t total = static_cast(5) + buf[2]; + if (total > maxAdu || len < total) { + return std::nullopt; // wait more bytes + } + return crcOk(buf, total) ? std::optional(total) : std::nullopt; + } + + if (fc == 0x05 || fc == 0x06) { + return len >= 8 ? std::optional(8) : std::nullopt; + } + if (fc == 0x16) { + return len >= 10 ? std::optional(10) : std::nullopt; + } + if (fc == 0x0F || fc == 0x10) { + if (len >= 8 && crcOk(buf, 8)) { + return std::optional(8); // response + } + if (len < 7) { + return std::nullopt; + } + const std::size_t req = static_cast(9) + buf[6]; + if (req > maxAdu || len < req) { + return std::nullopt; + } + return crcOk(buf, req) ? std::optional(req) : std::nullopt; + } + return len >= 8 ? std::optional(8) : std::nullopt; +} + +bool isModbusRtuHint(const QString& hint) +{ + const QString h = hint.trimmed().toLower(); + return h == QStringLiteral("modbus_rtu") || h == QStringLiteral("modbus-rtu") + || h == QStringLiteral("modbusrtu"); +} + +} // namespace softbus::message_bus::pipeline::protocol::modbusrtu diff --git a/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h b/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h new file mode 100644 index 0000000..645bd26 --- /dev/null +++ b/plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h @@ -0,0 +1,17 @@ +#pragma once + +#include +#include +#include + +#include + +namespace softbus::message_bus::pipeline::protocol::modbusrtu +{ + +std::uint16_t crc16(const std::uint8_t* data, std::size_t len); +bool crcOk(const std::uint8_t* adu, std::size_t aduLen); +std::optional predictAduLength(const std::uint8_t* buf, std::size_t len, std::size_t maxAdu); +bool isModbusRtuHint(const QString& hint); + +} // namespace softbus::message_bus::pipeline::protocol::modbusrtu diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp b/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.cpp similarity index 96% rename from src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp rename to plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.cpp index 2e4ef0c..2c69fda 100644 --- a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp +++ b/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.cpp @@ -1,10 +1,10 @@ -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h" +#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 "message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h" -#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" +#include "plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h" +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" #include #include @@ -227,7 +227,7 @@ public: bool supports(const QString& protocolHint) const override { - return softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu(protocolHint); + return softbus::message_bus::pipeline::protocol::modbusrtu::isModbusRtuHint(protocolHint); } std::unique_ptr createSession() override @@ -241,4 +241,9 @@ std::shared_ptr makeModbusRtuProtocolPlugin() return std::make_shared(); } +IProtocolPlugin* createModbusRtuProtocolPluginRaw() +{ + return new ModbusRtuProtocolPlugin(); +} + } // namespace softbus::core::plugin_system diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h b/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h similarity index 81% rename from src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h rename to plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h index d408b98..673e98f 100644 --- a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h +++ b/plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h @@ -8,5 +8,6 @@ namespace softbus::core::plugin_system { std::shared_ptr makeModbusRtuProtocolPlugin(); +IProtocolPlugin* createModbusRtuProtocolPluginRaw(); } // namespace softbus::core::plugin_system diff --git a/src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h b/plugins/protocols/modbusrtu/types/ModbusRtuPdu.h similarity index 100% rename from src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h rename to plugins/protocols/modbusrtu/types/ModbusRtuPdu.h diff --git a/src/core/plugin_system/IProtocolFramerPlugin.h b/src/core/plugin_system/IProtocolFramerPlugin.h new file mode 100644 index 0000000..bcf8c8b --- /dev/null +++ b/src/core/plugin_system/IProtocolFramerPlugin.h @@ -0,0 +1,38 @@ +#pragma once + +#include +#include + +#include +#include + +#include "core/memory/MemoryPool.h" +#include "message_bus/pipeline/PipelineContext.h" + +namespace softbus::core::plugin_system +{ + +class IProtocolFramerSession +{ +public: + virtual ~IProtocolFramerSession() = default; + + virtual void feedChunk(softbus::message_bus::pipeline::PipelineContext& in, + std::vector& out) = 0; + virtual void reset() = 0; + virtual void shutdown() = 0; +}; + +class IProtocolFramerPlugin +{ +public: + virtual ~IProtocolFramerPlugin() = default; + + virtual QString pluginId() const = 0; + virtual bool supports(const QString& protocolHint) const = 0; + virtual void bindMemoryPool(std::shared_ptr pool) = 0; + virtual std::unique_ptr createFramerSession(const QString& endpoint, + const QJsonObject& params) = 0; +}; + +} // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/PluginHost.cpp b/src/core/plugin_system/PluginHost.cpp new file mode 100644 index 0000000..03ba54e --- /dev/null +++ b/src/core/plugin_system/PluginHost.cpp @@ -0,0 +1,125 @@ +#include "core/plugin_system/PluginHost.h" + +#include + +#include "core/plugin_system/IProtocolFramerPlugin.h" +#include "core/plugin_system/IProtocolPlugin.h" + +namespace softbus::core::plugin_system +{ + +namespace +{ + +using GetAbiVersionFn = int (*)(); +using CreateProtocolPluginFn = IProtocolPlugin* (*)(); +using CreateFramerPluginFn = IProtocolFramerPlugin* (*)(); + +} // namespace + +PluginHost::PluginHost(std::shared_ptr manager) + : m_manager(std::move(manager)) +{ +} + +PluginHost::~PluginHost() +{ + unloadAll(); +} + +bool PluginHost::loadPlugin(const QString& path, QString* err) +{ + if (!m_manager) { + if (err) *err = QStringLiteral("plugin_manager_null"); + return false; + } + if (m_loaded.contains(path)) { + return true; + } + void* handle = dlopen(path.toUtf8().constData(), RTLD_NOW | RTLD_LOCAL); + if (!handle) { + if (err) *err = QString::fromUtf8(dlerror()); + return false; + } + auto* getAbi = reinterpret_cast(dlsym(handle, "softbus_plugin_abi_version")); + const int abi = getAbi ? getAbi() : 1; + if (abi != 1) { + if (err) *err = QStringLiteral("abi_mismatch"); + dlclose(handle); + return false; + } + + auto* createParser = reinterpret_cast(dlsym(handle, "softbus_create_protocol_plugin")); + QString protocolPluginId; + if (createParser) { + std::shared_ptr p(createParser()); + if (p) { + protocolPluginId = p->pluginId(); + m_manager->registerProtocolPlugin(p); + } + } + auto* createFramer = reinterpret_cast(dlsym(handle, "softbus_create_framer_plugin")); + QString framerPluginId; + if (createFramer) { + std::shared_ptr p(createFramer()); + if (p) { + framerPluginId = p->pluginId(); + m_manager->registerProtocolFramerPlugin(p); + } + } + LoadedPlugin lp; + lp.path = path; + lp.protocolPluginId = protocolPluginId; + lp.framerPluginId = framerPluginId; + lp.abiVersion = abi; + lp.handle = handle; + m_loaded.insert(path, lp); + return true; +} + +bool PluginHost::reloadPlugin(const QString& path, QString* err) +{ + QString localErr; + if (!unloadPlugin(path, &localErr)) { + // allow reload when not loaded + localErr.clear(); + } + const bool ok = loadPlugin(path, &localErr); + if (!ok && err) { + *err = localErr; + } + return ok; +} + +bool PluginHost::unloadPlugin(const QString& path, QString* err) +{ + auto it = m_loaded.find(path); + if (it == m_loaded.end()) { + if (err) *err = QStringLiteral("not_loaded"); + return false; + } + if (m_manager) { + if (!it.value().protocolPluginId.isEmpty()) { + m_manager->unregisterProtocolPlugin(it.value().protocolPluginId); + } + if (!it.value().framerPluginId.isEmpty()) { + m_manager->unregisterProtocolFramerPlugin(it.value().framerPluginId); + } + } + if (it.value().handle) { + dlclose(it.value().handle); + } + m_loaded.erase(it); + return true; +} + +void PluginHost::unloadAll() +{ + const auto keys = m_loaded.keys(); + for (const auto& k : keys) { + QString ignore; + (void)unloadPlugin(k, &ignore); + } +} + +} // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/PluginHost.h b/src/core/plugin_system/PluginHost.h new file mode 100644 index 0000000..ea59a13 --- /dev/null +++ b/src/core/plugin_system/PluginHost.h @@ -0,0 +1,39 @@ +#pragma once + +#include + +#include +#include + +#include "core/plugin_system/PluginManager.h" + +namespace softbus::core::plugin_system +{ + +class PluginHost +{ +public: + struct LoadedPlugin + { + QString path; + QString protocolPluginId; + QString framerPluginId; + int abiVersion{1}; + void* handle{nullptr}; + }; + + explicit PluginHost(std::shared_ptr manager); + ~PluginHost(); + + bool loadPlugin(const QString& path, QString* err = nullptr); + bool reloadPlugin(const QString& path, QString* err = nullptr); + bool unloadPlugin(const QString& path, QString* err = nullptr); + +private: + void unloadAll(); + + std::shared_ptr m_manager; + QHash m_loaded; +}; + +} // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/PluginManager.cpp b/src/core/plugin_system/PluginManager.cpp index 480bf17..70732a0 100644 --- a/src/core/plugin_system/PluginManager.cpp +++ b/src/core/plugin_system/PluginManager.cpp @@ -21,4 +21,32 @@ std::shared_ptr PluginManager::selectProtocolPlugin(const QStri return {}; } +void PluginManager::unregisterProtocolPlugin(const QString& pluginId) +{ + m_protocolPlugins.remove(pluginId); +} + +void PluginManager::registerProtocolFramerPlugin(std::shared_ptr plugin) +{ + if (!plugin) { + return; + } + m_protocolFramerPlugins.insert(plugin->pluginId(), std::move(plugin)); +} + +std::shared_ptr PluginManager::selectProtocolFramerPlugin(const QString& protocolHint) const +{ + for (auto it = m_protocolFramerPlugins.constBegin(); it != m_protocolFramerPlugins.constEnd(); ++it) { + if (it.value() && it.value()->supports(protocolHint)) { + return it.value(); + } + } + return {}; +} + +void PluginManager::unregisterProtocolFramerPlugin(const QString& pluginId) +{ + m_protocolFramerPlugins.remove(pluginId); +} + } // namespace softbus::core::plugin_system diff --git a/src/core/plugin_system/PluginManager.h b/src/core/plugin_system/PluginManager.h index ef450e7..6cb5965 100644 --- a/src/core/plugin_system/PluginManager.h +++ b/src/core/plugin_system/PluginManager.h @@ -5,6 +5,7 @@ #include #include +#include "core/plugin_system/IProtocolFramerPlugin.h" #include "core/plugin_system/IProtocolPlugin.h" namespace softbus::core::plugin_system @@ -17,9 +18,14 @@ public: void registerProtocolPlugin(std::shared_ptr plugin); std::shared_ptr selectProtocolPlugin(const QString& protocolHint) const; + void unregisterProtocolPlugin(const QString& pluginId); + void registerProtocolFramerPlugin(std::shared_ptr plugin); + std::shared_ptr selectProtocolFramerPlugin(const QString& protocolHint) const; + void unregisterProtocolFramerPlugin(const QString& pluginId); private: QHash> m_protocolPlugins; + QHash> m_protocolFramerPlugins; }; } // namespace softbus::core::plugin_system diff --git a/src/device_bus/manager/DeviceBusManager.cpp b/src/device_bus/manager/DeviceBusManager.cpp index 8cfc0a9..355a49f 100644 --- a/src/device_bus/manager/DeviceBusManager.cpp +++ b/src/device_bus/manager/DeviceBusManager.cpp @@ -3,6 +3,7 @@ #include #include +#include #include // #include @@ -10,7 +11,7 @@ #include "message_bus/ingress/DriverIngressAdapter.h" #include "message_bus/pipeline/PipelineEngine.h" #include "core/plugin_system/PluginManager.h" -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h" +#include "plugins/protocols/modbusrtu/parser/ModbusRtuProtocolPlugin.h" #include "device_bus/manager/SerialDeviceManager.h" #include "devices/DeviceTypes.h" #include "utils/logs/logging.h" @@ -57,7 +58,23 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig) if (!m_pluginManager) { m_pluginManager = std::make_shared(); } + 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; + } + } if (m_pipelineEngine) { m_pipelineEngine->setMemoryPool(m_memoryPool); m_pipelineEngine->loadRuntimeConfigFromJson(rootConfig.value(QStringLiteral("pipeline")).toObject()); diff --git a/src/device_bus/manager/DeviceBusManager.h b/src/device_bus/manager/DeviceBusManager.h index dbc1fa2..0f201dc 100644 --- a/src/device_bus/manager/DeviceBusManager.h +++ b/src/device_bus/manager/DeviceBusManager.h @@ -13,6 +13,7 @@ #include "core/memory/MemoryPool.h" #include "core/plugin_system/PluginManager.h" +#include "core/plugin_system/PluginHost.h" #include "device_bus/tree/DeviceTreeModel.h" #include "device_bus/registry/IDeviceRegistry.h" #include "devices/DeviceTypes.h" @@ -57,6 +58,7 @@ private: softbus::device_bus::registry::IDeviceRegistry* m_registry = nullptr; std::shared_ptr m_memoryPool; std::shared_ptr m_pluginManager; + std::unique_ptr m_pluginHost; std::unique_ptr m_pipelineEngine; std::unique_ptr m_ingressAdapter; std::unique_ptr m_egressRouter; diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index 3c95fc0..b9dbc4c 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -6,7 +6,8 @@ #include #include -#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" +#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" @@ -21,8 +22,7 @@ struct EndpointFramingState std::uint64_t nextSeq{1}; QString endpoint; std::shared_ptr pool; - softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig modbusCfg; - std::shared_ptr pass; + std::function&)> feed; std::function pushFramed; explicit EndpointFramingState(std::size_t cap) @@ -36,11 +36,6 @@ namespace void endpointFramerThread(std::shared_ptr st) { - using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer; - using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu; - - auto modbus = std::make_unique(st->pool, st->modbusCfg, st->endpoint); - while (st->running.load(std::memory_order_acquire)) { PipelineContext ch; if (!st->chunkQ.pop(ch)) { @@ -48,10 +43,8 @@ void endpointFramerThread(std::shared_ptr st) continue; } std::vector out; - if (protocolHintIsModbusRtu(ch.protocolHint)) { - modbus->feed(ch, out); - } else if (st->pass) { - st->pass->feed(ch, out); + if (st->feed) { + st->feed(ch, out); } for (auto& x : out) { x.frameSeq = st->nextSeq++; @@ -98,9 +91,9 @@ 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.modbus.maxAdu = static_cast( + m_rtConfig.modbusMaxAdu = static_cast( std::max(8, obj.value(QStringLiteral("modbusRtuMaxFrameBytes")).toInt(256))); - m_rtConfig.modbus.interFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0); + m_rtConfig.modbusInterFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0); } void PipelineEngine::start() @@ -115,12 +108,15 @@ void PipelineEngine::start() return; } - m_modbusFramer = - std::make_unique(m_pool, - m_rtConfig.modbus, - QString{}); 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)); + } // 如果启用并行管道 则创建帧队列和解析线程 如果启用有序出口 则创建合并队列和合并线程 // 并行管道是用来处理多个端点的数据流的 @@ -182,8 +178,11 @@ void PipelineEngine::stop() m_framedQ.reset(); m_mergeQ.reset(); - m_modbusFramer.reset(); m_passthroughFramer.reset(); + { + std::lock_guard lk(m_framerSessionMutex); + m_framerSessions.clear(); + } } // 功能 ：清空 ingress 队列 // 参数 ：无 @@ -266,8 +265,9 @@ void PipelineEngine::ensureEndpointFramer(const QString& endpoint) auto st = std::make_shared(qcap > 0 ? qcap : 4096); st->endpoint = endpoint; st->pool = m_pool; - st->modbusCfg = m_rtConfig.modbus; - st->pass = m_passthroughFramer; + st->feed = [this](PipelineContext& in, std::vector& out) { + this->dispatchFramer(in, out); + }; st->pushFramed = [this](PipelineContext c) { if (!m_framedQ) { return; @@ -286,17 +286,30 @@ void PipelineEngine::ensureEndpointFramer(const QString& endpoint) // 逻辑 ：根据协议提示 选择合适的帧解析器 进行帧解析 void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector& completeFramesOut) { - using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu; - // 根据协议提示 选择合适的帧解析器 进行帧解析 - - if (protocolHintIsModbusRtu(chunkIn.protocolHint)) { - if (m_modbusFramer) { - m_modbusFramer->feed(chunkIn, completeFramesOut); - + if (m_pluginManager) { + auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocolHint); + if (plugin) { + plugin->bindMemoryPool(m_pool); + const QString key = chunkIn.endpoint + QStringLiteral("|") + plugin->pluginId(); + std::lock_guard 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); + m_framerSessions.insert( + key, s ? std::shared_ptr(std::move(s)) : nullptr); + it = m_framerSessions.find(key); + } + if (it != m_framerSessions.end() && it.value()) { + it.value()->feedChunk(chunkIn, completeFramesOut); + return; + } } - } else if (m_passthroughFramer) { + } + if (m_passthroughFramer) { m_passthroughFramer->feed(chunkIn, completeFramesOut); - } } // 功能 ：分配帧序列号 diff --git a/src/message_bus/pipeline/PipelineEngine.h b/src/message_bus/pipeline/PipelineEngine.h index b29fa59..36c8e5b 100644 --- a/src/message_bus/pipeline/PipelineEngine.h +++ b/src/message_bus/pipeline/PipelineEngine.h @@ -11,11 +11,11 @@ #include #include "core/plugin_system/PluginManager.h" +#include "core/plugin_system/IProtocolFramerPlugin.h" #include "core/memory/LockFreeQueue.h" #include "core/memory/MemoryPool.h" #include "core/threading/MutexQueue.h" #include "message_bus/pipeline/PipelineContext.h" -#include "message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h" #include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h" namespace softbus::message_bus::pipeline @@ -30,7 +30,8 @@ struct PipelineRuntimeConfig std::size_t framedQueueMax{4096}; std::size_t maxReorderDepth{256}; - softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig modbus; + std::size_t modbusMaxAdu{256}; + int modbusInterFrameTimeoutMs{0}; }; struct EndpointFramingState; @@ -106,8 +107,9 @@ private: PipelineRuntimeConfig m_rtConfig; - std::unique_ptr m_modbusFramer; std::shared_ptr m_passthroughFramer; + std::mutex m_framerSessionMutex; + QHash> m_framerSessions; 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 890eeeb..bf47321 100644 --- a/src/message_bus/pipeline/protocol/ProtocolEnvelope.h +++ b/src/message_bus/pipeline/protocol/ProtocolEnvelope.h @@ -2,7 +2,7 @@ #include -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" +#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" namespace softbus::message_bus::pipeline::protocol { diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp deleted file mode 100644 index b1b00e9..0000000 --- a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.cpp +++ /dev/null @@ -1,201 +0,0 @@ -#include "message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h" - -#include -#include -#include -#include - -#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" -#include "utils/logs/logging.h" - -namespace softbus::message_bus::pipeline::stages::framers -{ - -ModbusRtuFramer::ModbusRtuFramer(std::shared_ptr pool, - ModbusRtuFramerConfig cfg, - QString exclusiveEndpoint) - : m_pool(std::move(pool)) - , m_cfg(cfg) - , m_exclusiveEndpoint(std::move(exclusiveEndpoint)) -{ -} - -// 功能 ：追加数据块 将数据块追加到会话中 通过时间戳记录最后接收时间 时间间隔超过最大帧间隔时间 则清空会话 如果超过最大ADU长度 则记录溢出 并记录溢出数量 -void ModbusRtuFramer::appendChunk(Session& session, const std::uint8_t* p, std::size_t n) -{ - if (!p || n == 0) { - return; - } - session.lastRx = std::chrono::steady_clock::now(); - if (session.buf.size() < m_cfg.maxAdu) { - session.buf.resize(m_cfg.maxAdu); - } - if (session.len + n > m_cfg.maxAdu) { - ++m_cnt.overflow; - session.len = 0; - LOG_WARNING() << "ModbusRtuFramer: buffer_overflow maxAdu=" << m_cfg.maxAdu; - return; - } - // `p/n` 来自 chunkIn.payload.bytes()，也就是串口读回的原始字节块（当前函数会再次拷贝进 session.buf） - const std::size_t beforeLen = session.len; - std::ostringstream oss; - oss << std::hex << std::setfill('0'); - const std::size_t show = std::min(n, static_cast(16)); - for (std::size_t i = 0; i < show; ++i) { - oss << std::setw(2) << static_cast(p[i]); - if (i + 1 < show) { - oss << ' '; - } - } - if (n > show) { - oss << " ..."; - } - const QString hexPreview = QString::fromStdString(oss.str()); - // LOG_INFO() << "ModbusRtuFramer: appendChunk n=" << n << " session.len(before)=" << beforeLen - // << " p=" << static_cast(p) << " bytes(16)=" << hexPreview; - std::memcpy(session.buf.data() + session.len, p, n); - session.len += n; - m_cnt.bytesIn += n; -} -// 功能 ：尝试消费缓冲区 如果时间间隔超过最大帧间隔时间 则清空会话 如果超过最大ADU长度 则记录溢出 并记录溢出数量 -void ModbusRtuFramer::tryConsumeBuffer(Session& session, - const QString& endpoint, - QString traceId, - int deviceId, - const QString& stableKey, - const QString& protocolHint, - QDateTime ts, - BusDirection dir, - std::vector& out, - QString* framerErr) -{ - (void)framerErr; - - if (m_cfg.interFrameTimeoutMs > 0 && session.len > 0) { - const auto now = std::chrono::steady_clock::now(); - const auto ms = std::chrono::duration_cast(now - session.lastRx).count(); - if (ms >= m_cfg.interFrameTimeoutMs) { - session.len = 0; - } - } - - while (session.len > 0) { - const std::uint8_t* buf = session.buf.data(); - auto pred = - predictModbusRtuAduLength(buf, session.len, m_cfg.maxAdu); - if (!pred) { - break; - } - if (session.len < *pred) { - break; - } - - // std::ostringstream oss; - // oss << std::hex << std::setfill('0'); - // for (std::size_t i = 0; i < session.len; ++i) { - // oss << std::setw(2) << static_cast(buf[i]); - // if (i + 1 < session.len) { - // oss << ' '; - // } - // } - // const QString hexPreview = QString::fromStdString(oss.str()); - // LOG_INFO() << "ModbusRtuFramer: tryConsumeBuffer buf(16)=" << hexPreview << " session.len=" << session.len << " pred=" << *pred ; - - if (modbusRtuCrcOk(buf, *pred)) { - auto blk = m_pool->allocate(*pred); - if (!blk || !blk->data || blk->capacity < *pred) { - if (framerErr) { - *framerErr = QStringLiteral("pool_exhausted"); - } - LOG_WARNING() << "ModbusRtuFramer: pool_exhausted endpoint=" << endpoint; - return; - } - std::memcpy(blk->data, buf, *pred); - blk->size = *pred; - - softbus::message_bus::pipeline::PipelineContext ctx; - ctx.deviceId = deviceId; - ctx.stableKey = stableKey; - ctx.endpoint = endpoint; - ctx.timestamp = ts; - ctx.protocolHint = protocolHint; - ctx.traceId = traceId; - ctx.direction = dir; - ctx.frameKind = softbus::message_bus::pipeline::FrameKind::CompleteFrame; - ctx.payloadSize = *pred; - ctx.payload.block = std::move(blk); - ctx.payload.offset = 0; - ctx.payload.length = *pred; - - out.push_back(std::move(ctx)); - ++m_cnt.framesEmitted; - - const std::size_t rest = session.len - *pred; - if (rest > 0) { - std::memmove(session.buf.data(), buf + *pred, rest); - } - session.len = rest; - // LOG_INFO() << "ModbusRtuFramer success to consume buffer rest=" << rest; - continue; - } - - ++m_cnt.crcFail; - ++m_cnt.resyncBytes; - // LOG_WARNING() << "ModbusRtuFramer: crc_mismatch resync_byte endpoint=" << endpoint - // << "traceId=" << traceId << "buf_len=" << session.len; - - if (session.len <= 1) { - session.len = 0; - break; - } - std::memmove(session.buf.data(), buf + 1, session.len - 1); - --session.len; - } -} - -void ModbusRtuFramer::feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, - std::vector& completeFramesOut) -{ - if (!m_pool || !chunkIn.payload.valid()) { - return; - } - if (!protocolHintIsModbusRtu(chunkIn.protocolHint)) { - return; - } - - const std::uint8_t* p = chunkIn.payload.bytes(); - const std::size_t n = chunkIn.payloadSize ? chunkIn.payloadSize : chunkIn.payload.length; - if (!p || n == 0) { - return; - } - - Session* session = nullptr; - if (!m_exclusiveEndpoint.isEmpty()) { - session = &m_exclusiveSession; - } else { - session = &m_sessions[chunkIn.endpoint]; - } - - appendChunk(*session, p, n); - - QString err; - tryConsumeBuffer(*session, - chunkIn.endpoint, - chunkIn.traceId, - chunkIn.deviceId, - chunkIn.stableKey, - chunkIn.protocolHint, - chunkIn.timestamp, - chunkIn.direction, - completeFramesOut, - &err); - if (!err.isEmpty()) { - chunkIn.framerError = err; - } - - chunkIn.payload = {}; - chunkIn.payloadSize = 0; - chunkIn.frameKind = softbus::message_bus::pipeline::FrameKind::StreamChunk; -} - -} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h deleted file mode 100644 index 8532f3d..0000000 --- a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFramer.h +++ /dev/null @@ -1,79 +0,0 @@ -#pragma once - -#include -#include -#include -#include -#include -#include - -#include -#include - -#include "core/memory/MemoryPool.h" -#include "message_bus/pipeline/PipelineContext.h" -#include "message_bus/pipeline/stages/1_framers/IFramer.h" - -namespace softbus::message_bus::pipeline::stages::framers -{ - -struct ModbusRtuFramerConfig -{ - std::size_t maxAdu{256}; - int interFrameTimeoutMs{0}; -}; - -struct ModbusRtuFramerCounters -{ - std::atomic bytesIn{0}; // 输入字节数 - std::atomic framesEmitted{0}; // 输出帧数 - std::atomic resyncBytes{0}; // 重同步字节数 - std::atomic crcFail{0}; // CRC校验失败数 - std::atomic overflow{0}; // 溢出数 -}; - -/// 单 worker、多 endpoint：使用内部 QHash（单线程访问 Session，无锁）。 -/// 并行成帧：构造时传入 `exclusiveEndpoint`，仅重组该串口字节流（每 endpoint 一实例）。 -class ModbusRtuFramer final : public IFramer -{ -public: - ModbusRtuFramer(std::shared_ptr pool, - ModbusRtuFramerConfig cfg = {}, - QString exclusiveEndpoint = {}); - - void feed(softbus::message_bus::pipeline::PipelineContext& chunkIn, - std::vector& completeFramesOut) override; - - const ModbusRtuFramerCounters& counters() const { return m_cnt; } - ModbusRtuFramerCounters& counters() { return m_cnt; } - void setConfig(ModbusRtuFramerConfig cfg) { m_cfg = cfg; } - -private: - struct Session - { - std::vector buf; - std::size_t len{0}; - std::chrono::steady_clock::time_point lastRx{std::chrono::steady_clock::now()}; - }; - - void appendChunk(Session& session, const std::uint8_t* p, std::size_t n); - void tryConsumeBuffer(Session& session, - const QString& endpoint, - QString traceId, - int deviceId, - const QString& stableKey, - const QString& protocolHint, - QDateTime ts, - BusDirection dir, - std::vector& out, - QString* framerErr); - - std::shared_ptr m_pool; - ModbusRtuFramerConfig m_cfg; - ModbusRtuFramerCounters m_cnt; - QString m_exclusiveEndpoint; - QHash m_sessions; - Session m_exclusiveSession; -}; - -} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp deleted file mode 100644 index 65518cd..0000000 --- a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp +++ /dev/null @@ -1,125 +0,0 @@ -#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" - -#include - -namespace softbus::message_bus::pipeline::stages::framers -{ - -std::uint16_t modbusRtuCrc16(const std::uint8_t* data, std::size_t len) -{ - std::uint16_t crc = 0xFFFF; - for (std::size_t i = 0; i < len; ++i) { - crc ^= data[i]; - for (int k = 0; k < 8; ++k) { - if (crc & 1) { - crc = static_cast((crc >> 1) ^ 0xA001); - } else { - crc = static_cast(crc >> 1); - } - } - } - return crc; -} - -// 功能 ：校验CRC 如果CRC校验不正确 则返回false 如果CRC校验正确 则返回true -// 参数 ：adu 数据缓冲区 aduLen 数据长度 -// 返回值 ：如果CRC校验不正确 则返回false 如果CRC校验正确 则返回true -// 逻辑：如果数据缓冲区为空 则返回false 如果数据长度小于4 则返回false 如果数据长度大于等于4 则返回true -// 如果数据长度大于等于4 则返回true -// 如果数据长度大于等于4 则返回true -bool modbusRtuCrcOk(const std::uint8_t* adu, std::size_t aduLen) -{ - if (!adu || aduLen < 4) { - return false; - } - const std::size_t body = aduLen - 2; - const std::uint16_t got = - static_cast(adu[body] | (static_cast(adu[body + 1]) << 8)); - const std::uint16_t want = modbusRtuCrc16(adu, body); - return got == want; -} - -static bool fcIsReadRange(std::uint8_t fc) -{ - return fc == 0x01 || fc == 0x02 || fc == 0x03 || fc == 0x04; -} - -// 功能 ：预测Modbus RTU帧长度 如果长度不足 则返回nullopt 如果长度足够 则返回长度 -// 参数 ：buf 数据缓冲区 len 数据长度 maxAdu 最大ADU长度 -// 返回值 ：如果长度不足 则返回nullopt 如果长度足够 则返回长度 -// 逻辑： unit 是设备地址 fc 是功能码 如果fc是异常码 则返回5 如果fc是读指令 则返回8 如果fc是写指令 则返回8 如果fc是其他指令 则返回nullopt -// 如果fc是读指令 则检查buf长度是否大于等于8 并且CRC校验是否正确 如果正确 则返回8 -// 如果fc是写指令 则检查buf长度是否大于等于8 并且CRC校验是否正确 如果正确 则返回8 - -std::optional predictModbusRtuAduLength(const std::uint8_t* buf, - std::size_t len, - std::size_t maxAdu) -{ - if (len < 2 || !buf) { - return std::nullopt; - } - const std::uint8_t unit = buf[0]; - (void)unit; - const std::uint8_t fc = buf[1]; - - if (fc & 0x80) { - return len < 5 ? std::nullopt : std::optional(5); - } - - if (fcIsReadRange(fc)) { - if (len >= 8 && modbusRtuCrcOk(buf, 8)) { - return std::optional(8); - } - if (len < 3) { - return std::nullopt; - } - const std::size_t byteCount = buf[2]; - const std::size_t total = 5 + byteCount; - if (total > maxAdu) { - return std::nullopt; - } - if (len < total) { - return std::nullopt; - } - if (modbusRtuCrcOk(buf, total) ) { - return total; - } - return std::optional(8); - } - - if (fc == 0x05 || fc == 0x06) { - return std::optional(8); - } - - if (fc == 0x16) { - return std::optional(10); - } - - if (fc == 0x0F || fc == 0x10) { - if (len >= 8 && modbusRtuCrcOk(buf, 8)) { - return std::optional(8); - } - if (len >= 7) { - const std::uint8_t bc = buf[6]; - const std::size_t req = 9 + static_cast(bc); - if (bc <= 242 && req <= maxAdu && req >= 9) { - return req; - } - } - return len >= 2 ? std::optional(8) : std::nullopt; - } - - if (len < 6) { - return std::nullopt; - } - return std::optional(8); -} - -bool protocolHintIsModbusRtu(const QString& hint) -{ - const QString h = hint.trimmed().toLower(); - return h == QStringLiteral("modbus_rtu") || h == QStringLiteral("modbus-rtu") - || h == QStringLiteral("modbusrtu"); -} - -} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h b/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h deleted file mode 100644 index 71c4609..0000000 --- a/src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h +++ /dev/null @@ -1,24 +0,0 @@ -#pragma once -// 功能 ：定义Modbus RTU帧解析器 包含CRC校验、帧预测、协议提示判断 -#include -#include -#include - -#include - -namespace softbus::message_bus::pipeline::stages::framers -{ - -std::uint16_t modbusRtuCrc16(const std::uint8_t* data, std::size_t len); - -/// 校验 ADU（含尾端 CRC LE）；`aduLen` 为整帧长度（unit..crc_hi）。 -bool modbusRtuCrcOk(const std::uint8_t* adu, std::size_t aduLen); - -/// 根据已缓冲前缀预测整帧 ADU 长度（含 2 字节 CRC）；未知或不足返回 nullopt。 -std::optional predictModbusRtuAduLength(const std::uint8_t* buf, - std::size_t len, - std::size_t maxAdu); - -bool protocolHintIsModbusRtu(const QString& hint); - -} // namespace softbus::message_bus::pipeline::stages::framers diff --git a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp index 2e41d36..d8a65dc 100644 --- a/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp +++ b/src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp @@ -16,7 +16,7 @@ #include -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" +#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" #include "message_bus/pipeline/protocol/ProtocolEnvelope.h" #include "utils/logs/logging.h" diff --git a/tests/modbus_rtu_framer_test.cpp b/tests/modbus_rtu_framer_test.cpp index 619a901..9b600a7 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 "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h" +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" #include @@ -11,14 +11,14 @@ using softbus::message_bus::pipeline::FrameKind; using softbus::message_bus::pipeline::PipelineContext; using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer; using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig; -using softbus::message_bus::pipeline::stages::framers::modbusRtuCrc16; +using softbus::message_bus::pipeline::protocol::modbusrtu::crc16; namespace { void appendCrcLe(std::vector& v) { - const std::uint16_t c = modbusRtuCrc16(v.data(), v.size()); + const std::uint16_t c = crc16(v.data(), v.size()); v.push_back(static_cast(c & 0xFF)); v.push_back(static_cast(c >> 8)); } diff --git a/tests/modbus_rtu_framing_predict_test.cpp b/tests/modbus_rtu_framing_predict_test.cpp new file mode 100644 index 0000000..3080f5d --- /dev/null +++ b/tests/modbus_rtu_framing_predict_test.cpp @@ -0,0 +1,57 @@ +#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h" + +#include + +using softbus::message_bus::pipeline::protocol::modbusrtu::predictAduLength; +using softbus::message_bus::pipeline::protocol::modbusrtu::crc16; + +namespace +{ +void appendCrc(QByteArray& b) +{ + const auto c = crc16(reinterpret_cast(b.constData()), static_cast(b.size())); + b.append(static_cast(c & 0xFF)); + b.append(static_cast((c >> 8) & 0xFF)); +} +} // namespace + +class ModbusRtuFramingPredictTest : public QObject +{ + Q_OBJECT +private slots: + void readRequestWaitsUntilEightBytes(); + void readResponseNeedsWholeAndCrc(); +}; + +void ModbusRtuFramingPredictTest::readRequestWaitsUntilEightBytes() +{ + QByteArray partial; + partial.append('\x01'); + partial.append('\x03'); + partial.append('\x00'); + partial.append('\x00'); + partial.append('\x00'); + partial.append('\x0A'); + partial.append('\xC5'); // missing last crc byte + auto pred = predictAduLength(reinterpret_cast(partial.constData()), + static_cast(partial.size()), 256); + QVERIFY(!pred.has_value()); +} + +void ModbusRtuFramingPredictTest::readResponseNeedsWholeAndCrc() +{ + QByteArray resp; + resp.append('\x01'); + resp.append('\x03'); + resp.append('\x02'); + resp.append('\x12'); + resp.append('\x34'); + appendCrc(resp); + auto pred = predictAduLength(reinterpret_cast(resp.constData()), + static_cast(resp.size()), 256); + QVERIFY(pred.has_value()); + QCOMPARE(static_cast(*pred), resp.size()); +} + +QTEST_MAIN(ModbusRtuFramingPredictTest) +#include "modbus_rtu_framing_predict_test.moc" diff --git a/tests/modbus_rtu_pdu_codec_test.cpp b/tests/modbus_rtu_pdu_codec_test.cpp index fe7b0bb..cf667a9 100644 --- a/tests/modbus_rtu_pdu_codec_test.cpp +++ b/tests/modbus_rtu_pdu_codec_test.cpp @@ -1,4 +1,4 @@ -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h" +#include "plugins/protocols/modbusrtu/codec/ModbusRtuCodec.h" #include diff --git a/tests/protocol_parse_filter_test.cpp b/tests/protocol_parse_filter_test.cpp index 26f0124..5431701 100644 --- a/tests/protocol_parse_filter_test.cpp +++ b/tests/protocol_parse_filter_test.cpp @@ -6,7 +6,7 @@ #include #include -#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h" +#include "plugins/protocols/modbusrtu/types/ModbusRtuPdu.h" #include "message_bus/pipeline/protocol/ProtocolEnvelope.h" using softbus::message_bus::pipeline::BusDirection;