From 0cf0713dc5385d997dbd79310fad9f2692211768 Mon Sep 17 00:00:00 2001 From: flower_linux <737666956@qq.com> Date: Thu, 11 Jun 2026 19:00:49 +0800 Subject: [PATCH] phase 2 --- .gitignore | 5 + CMakeLists.txt | 7 + README.md | 7 +- config/edge.json | 9 + config/point_table.json | 46 ++++ config/point_table.yaml | 29 +++ config/topic_rule.json | 7 + config/topic_rule.yaml | 6 + docs/command_flow.md | 29 +++ docs/point_table_design.md | 20 ++ .../include/softbus_api/ExecuteCommand.h | 1 + src/softbus_api/src/ExecuteCommand.cpp | 17 ++ src/softbus_bus/CMakeLists.txt | 8 + src/softbus_bus/include/softbus_bus/Bus.h | 41 ++++ .../include/softbus_bus/BusMessage.h | 31 +++ src/softbus_bus/src/Bus.cpp | 107 ++++++++++ src/softbus_command/CMakeLists.txt | 9 + .../include/softbus_command/CommandContext.h | 34 +++ .../include/softbus_command/CommandService.h | 39 ++++ src/softbus_command/src/CommandContext.cpp | 21 ++ src/softbus_command/src/CommandService.cpp | 163 ++++++++++++++ src/softbus_core/CMakeLists.txt | 1 + .../include/softbus_core/models/ObjectModel.h | 9 + .../softbus_core/models/SoftbusTypes.h | 73 +++++++ src/softbus_core/src/models/ObjectModel.cpp | 21 ++ src/softbus_core/src/models/SoftbusTypes.cpp | 128 +++++++++++ src/softbus_daemon/CMakeLists.txt | 10 +- src/softbus_daemon/src/CoreService.cpp | 106 +++++++--- src/softbus_daemon/src/main.cpp | 8 +- src/softbus_edge/CMakeLists.txt | 11 +- .../include/softbus_edge/config/EdgeConfig.h | 18 ++ .../softbus_edge/egress/CommandExecutor.h | 26 +++ .../softbus_edge/egress/ShadowChannelStore.h | 21 ++ .../softbus_edge/egress/UplinkClient.h | 3 + src/softbus_edge/src/CommandExecutor.cpp | 69 ++++++ src/softbus_edge/src/EdgeConfig.cpp | 28 +++ src/softbus_edge/src/ShadowChannelStore.cpp | 32 +++ src/softbus_edge/src/UplinkClient.cpp | 10 + src/softbus_edge/src/main.cpp | 91 ++++++-- src/softbus_monitor/CMakeLists.txt | 8 + .../softbus_monitor/HeartbeatMonitor.h | 30 +++ src/softbus_monitor/src/HeartbeatMonitor.cpp | 70 ++++++ src/softbus_opcua/CMakeLists.txt | 2 +- .../include/softbus_opcua/UaEventBridge.h | 7 + .../include/softbus_opcua/UaMethodBinder.h | 4 +- .../include/softbus_opcua/UaModelBinder.h | 9 + src/softbus_opcua/src/UaEventBridge.cpp | 19 ++ src/softbus_opcua/src/UaMethodBinder.cpp | 7 +- src/softbus_opcua/src/UaModelBinder.cpp | 63 ++++-- src/softbus_pipeline/CMakeLists.txt | 10 + .../include/softbus_pipeline/PipelineEngine.h | 31 +++ src/softbus_pipeline/src/PipelineEngine.cpp | 108 ++++++++++ src/softbus_registry/CMakeLists.txt | 9 + .../include/softbus_registry/PointRegistry.h | 42 ++++ .../softbus_registry/PointTableEntry.h | 41 ++++ src/softbus_registry/src/PointRegistry.cpp | 199 ++++++++++++++++++ src/softbus_registry/src/PointTableEntry.cpp | 45 ++++ src/softbus_storage/CMakeLists.txt | 8 + .../include/softbus_storage/StorageService.h | 33 +++ src/softbus_storage/src/StorageService.cpp | 70 ++++++ src/softbus_transport/CMakeLists.txt | 3 +- .../softbus_transport/TcpUplinkClient.h | 3 + .../softbus_transport/TcpUplinkServer.h | 3 + .../softbus_transport/TransportBusBridge.h | 21 ++ .../include/softbus_transport/UplinkWire.h | 7 + src/softbus_transport/src/TcpUplinkClient.cpp | 36 ++++ src/softbus_transport/src/TcpUplinkServer.cpp | 8 + .../src/TransportBusBridge.cpp | 70 ++++++ src/softbus_transport/src/UplinkWire.cpp | 28 +++ tests/integration/valve_control_loop.sh | 26 +++ tools/CMakeLists.txt | 11 + tools/cmd_sender.cpp | 45 ++++ tools/topic_watcher.cpp | 15 ++ 73 files changed, 2297 insertions(+), 95 deletions(-) create mode 100644 config/edge.json create mode 100644 config/point_table.json create mode 100644 config/point_table.yaml create mode 100644 config/topic_rule.json create mode 100644 config/topic_rule.yaml create mode 100644 docs/command_flow.md create mode 100644 docs/point_table_design.md create mode 100644 src/softbus_bus/CMakeLists.txt create mode 100644 src/softbus_bus/include/softbus_bus/Bus.h create mode 100644 src/softbus_bus/include/softbus_bus/BusMessage.h create mode 100644 src/softbus_bus/src/Bus.cpp create mode 100644 src/softbus_command/CMakeLists.txt create mode 100644 src/softbus_command/include/softbus_command/CommandContext.h create mode 100644 src/softbus_command/include/softbus_command/CommandService.h create mode 100644 src/softbus_command/src/CommandContext.cpp create mode 100644 src/softbus_command/src/CommandService.cpp create mode 100644 src/softbus_core/include/softbus_core/models/SoftbusTypes.h create mode 100644 src/softbus_core/src/models/SoftbusTypes.cpp create mode 100644 src/softbus_edge/include/softbus_edge/config/EdgeConfig.h create mode 100644 src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h create mode 100644 src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h create mode 100644 src/softbus_edge/src/CommandExecutor.cpp create mode 100644 src/softbus_edge/src/EdgeConfig.cpp create mode 100644 src/softbus_edge/src/ShadowChannelStore.cpp create mode 100644 src/softbus_monitor/CMakeLists.txt create mode 100644 src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h create mode 100644 src/softbus_monitor/src/HeartbeatMonitor.cpp create mode 100644 src/softbus_pipeline/CMakeLists.txt create mode 100644 src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h create mode 100644 src/softbus_pipeline/src/PipelineEngine.cpp create mode 100644 src/softbus_registry/CMakeLists.txt create mode 100644 src/softbus_registry/include/softbus_registry/PointRegistry.h create mode 100644 src/softbus_registry/include/softbus_registry/PointTableEntry.h create mode 100644 src/softbus_registry/src/PointRegistry.cpp create mode 100644 src/softbus_registry/src/PointTableEntry.cpp create mode 100644 src/softbus_storage/CMakeLists.txt create mode 100644 src/softbus_storage/include/softbus_storage/StorageService.h create mode 100644 src/softbus_storage/src/StorageService.cpp create mode 100644 src/softbus_transport/include/softbus_transport/TransportBusBridge.h create mode 100644 src/softbus_transport/src/TransportBusBridge.cpp create mode 100755 tests/integration/valve_control_loop.sh create mode 100644 tools/CMakeLists.txt create mode 100644 tools/cmd_sender.cpp create mode 100644 tools/topic_watcher.cpp diff --git a/.gitignore b/.gitignore index fa15140..2f5eb3b 100644 --- a/.gitignore +++ b/.gitignore @@ -87,6 +87,11 @@ qrc_*.cpp __pycache__/ .cache/ +# ------------------------------------------------------------------------------ +# Runtime storage (softbus_storage) +# ------------------------------------------------------------------------------ +/data/ + # ------------------------------------------------------------------------------ # Logs & temp # ------------------------------------------------------------------------------ diff --git a/CMakeLists.txt b/CMakeLists.txt index 99bb146..4d597ff 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -17,12 +17,19 @@ include(SoftbusThirdParty) include(SoftbusPackage) add_subdirectory(src/softbus_core) +add_subdirectory(src/softbus_registry) +add_subdirectory(src/softbus_bus) add_subdirectory(src/softbus_transport) add_subdirectory(src/softbus_api) +add_subdirectory(src/softbus_command) +add_subdirectory(src/softbus_pipeline) +add_subdirectory(src/softbus_monitor) +add_subdirectory(src/softbus_storage) add_subdirectory(src/softbus_opcua) add_subdirectory(src/plugins/protocol_modbus) add_subdirectory(src/plugins/protocol_canopen) add_subdirectory(src/softbus_edge) add_subdirectory(src/softbus_daemon) +add_subdirectory(tools) message(STATUS "Softbus workspace configured for ${SOFTBUS_PLATFORM_NAME}") diff --git a/README.md b/README.md index 66054b5..5c07b08 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,7 @@ build/lib/protocol_canopen.so # CANopen 协议插件 ```bash ./build/bin/softbus_daemon \ --config config/daemon_profile.json \ - --model config/softbus_model.json + --point-table config/point_table.json ``` 命令行参数： @@ -66,7 +66,8 @@ build/lib/protocol_canopen.so # CANopen 协议插件 | 参数 | 默认值 | 说明 | |------|--------|------| | `--config ` | `config/daemon_profile.json` | 守护进程运行配置 | -| `--model ` | `config/softbus_model.json` | 对象模型定义 | +| `--point-table ` | `config/point_table.json` | 点表（含 ValveOpening 等） | +| `--model ` | 同 `--point-table` | 兼容旧参数名 | | `--self-test` | — | 启动后自动下发一条 ExecuteCommand 写值测试 | 守护进程启动后会： @@ -111,7 +112,7 @@ opc.tcp://localhost:4840 } ``` -`config/softbus_model.json` 定义站点/设备/点位层级，守护进程据此构建 OPC UA 地址空间。 +`config/point_table.json` 定义设备/点位/主题/OPC UA 映射，由 `softbus_registry` 加载并驱动 OPC UA 地址空间与控制闭环。 ## 典型联调流程 diff --git a/config/edge.json b/config/edge.json new file mode 100644 index 0000000..c81a4e3 --- /dev/null +++ b/config/edge.json @@ -0,0 +1,9 @@ +{ + "edge": { + "deviceId": "EdgeDevice_01", + "daemonHost": "127.0.0.1", + "daemonPort": 9000, + "heartbeatPeriodMs": 1000, + "pointTable": "config/point_table.json" + } +} diff --git a/config/point_table.json b/config/point_table.json new file mode 100644 index 0000000..3a59970 --- /dev/null +++ b/config/point_table.json @@ -0,0 +1,46 @@ +{ + "site": "DemoSite", + "system": "DemoSystem", + "asset": "DemoAsset", + "devices": [ + { + "deviceId": "EdgeDevice_01", + "stableDeviceKey": "EdgeDevice_01", + "protocol": "modbus", + "points": [ + { + "pointId": "MotorTemperature", + "name": "MotorTemperature", + "displayName": "电机温度", + "signalType": "AI", + "physicalChannel": "AI_Channel_01", + "register": 40001, + "unit": "℃", + "scale": 0.1, + "range": [0, 200], + "access": "read", + "writable": false, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature", + "dataTopic": "data/EdgeDevice_01/MotorTemperature" + }, + { + "pointId": "ValveOpening", + "name": "ValveOpening", + "displayName": "阀门开度", + "signalType": "AO", + "physicalChannel": "AO_Channel_01", + "register": 40002, + "unit": "%", + "scale": 1.0, + "range": [0, 100], + "access": "read_write", + "writable": true, + "opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening", + "cmdTopic": "control/cmd/EdgeDevice_01/ValveOpening/set", + "ackTopic": "control/ack/EdgeDevice_01/ValveOpening", + "dataTopic": "data/EdgeDevice_01/ValveOpening" + } + ] + } + ] +} diff --git a/config/point_table.yaml b/config/point_table.yaml new file mode 100644 index 0000000..91b8b34 --- /dev/null +++ b/config/point_table.yaml @@ -0,0 +1,29 @@ +# 点表配置（运行时加载 point_table.json；本文件为设计文档镜像） +site: DemoSite +system: DemoSystem +asset: DemoAsset +devices: + - deviceId: EdgeDevice_01 + stableDeviceKey: EdgeDevice_01 + protocol: modbus + points: + - pointId: MotorTemperature + displayName: 电机温度 + signalType: AI + physicalChannel: AI_Channel_01 + register: 40001 + unit: "℃" + range: [0, 200] + access: read + dataTopic: data/EdgeDevice_01/MotorTemperature + - pointId: ValveOpening + displayName: 阀门开度 + signalType: AO + physicalChannel: AO_Channel_01 + register: 40002 + unit: "%" + range: [0, 100] + access: read_write + cmdTopic: control/cmd/EdgeDevice_01/ValveOpening/set + ackTopic: control/ack/EdgeDevice_01/ValveOpening + dataTopic: data/EdgeDevice_01/ValveOpening diff --git a/config/topic_rule.json b/config/topic_rule.json new file mode 100644 index 0000000..704f88b --- /dev/null +++ b/config/topic_rule.json @@ -0,0 +1,7 @@ +{ + "data": "data/{deviceId}/{pointId}", + "command": "control/cmd/{deviceId}/{pointId}/set", + "ack": "control/ack/{deviceId}/{pointId}", + "event": "event/{deviceId}/{eventType}", + "heartbeat": "system/heartbeat/{deviceId}" +} diff --git a/config/topic_rule.yaml b/config/topic_rule.yaml new file mode 100644 index 0000000..3ab30fc --- /dev/null +++ b/config/topic_rule.yaml @@ -0,0 +1,6 @@ +topicRules: + data: "data/{deviceId}/{pointId}" + command: "control/cmd/{deviceId}/{pointId}/set" + ack: "control/ack/{deviceId}/{pointId}" + event: "event/{deviceId}/{eventType}" + heartbeat: "system/heartbeat/{deviceId}" diff --git a/docs/command_flow.md b/docs/command_flow.md new file mode 100644 index 0000000..342725d --- /dev/null +++ b/docs/command_flow.md @@ -0,0 +1,29 @@ +# 控制命令闭环 + +## 下行链路 + +``` +OPC UA Client / --self-test + → UaMethodBinder::ExecuteCommand + → CommandService::submit() + → registry 点表校验 + 范围检查 + → bus.publish(control/cmd/{device}/{point}/set) + → TransportBusBridge → TcpUplinkServer → edge + → CommandExecutor → ShadowChannelStore (mock AO) + → SoftbusAck → bus.publish(control/ack/...) + → CommandService ACK 匹配 → Succeeded +``` + +## 上行反馈 + +``` +ShadowChannelStore 更新 + → edge 构造 RawPacket 上行 + → PipelineEngine 解析 → UaModelBinder + → bus.publish(data/{device}/{point}) + → OPC UA Variable 更新 +``` + +## 命令状态 + +`Created → Validated → Dispatched → Succeeded / Failed / Timeout` diff --git a/docs/point_table_design.md b/docs/point_table_design.md new file mode 100644 index 0000000..78e6152 --- /dev/null +++ b/docs/point_table_design.md @@ -0,0 +1,20 @@ +# 点表设计 + +配置文件：`config/point_table.json`（YAML 镜像：`config/point_table.yaml`） + +## 映射关系 + +``` +ValveOpening + ↔ AO_Channel_01 (physicalChannel) + ↔ ns=1;s=.../ValveOpening (opcuaNodeId) + ↔ control/cmd/EdgeDevice_01/ValveOpening/set (cmdTopic) + ↔ control/ack/EdgeDevice_01/ValveOpening (ackTopic) + ↔ data/EdgeDevice_01/ValveOpening (dataTopic) +``` + +由 `softbus_registry::PointRegistry` 加载并建立三向索引：`objectRef` / `topic` / `opcuaNodeId`。 + +## 主题规则 + +`config/topic_rule.json` 定义默认主题模板，点表可覆盖单点 `cmdTopic`/`dataTopic`。 diff --git a/src/softbus_api/include/softbus_api/ExecuteCommand.h b/src/softbus_api/include/softbus_api/ExecuteCommand.h index 1befa97..c9a7e7d 100644 --- a/src/softbus_api/include/softbus_api/ExecuteCommand.h +++ b/src/softbus_api/include/softbus_api/ExecuteCommand.h @@ -6,6 +6,7 @@ namespace softbus::api { struct ExecuteCommand { + std::string requestId; std::string traceId; std::string objectRef; std::string command; diff --git a/src/softbus_api/src/ExecuteCommand.cpp b/src/softbus_api/src/ExecuteCommand.cpp index e28bbab..abdc09e 100644 --- a/src/softbus_api/src/ExecuteCommand.cpp +++ b/src/softbus_api/src/ExecuteCommand.cpp @@ -1,10 +1,13 @@ #include +#include + namespace softbus::api { ExecuteCommand ExecuteCommand::fromJson(const nlohmann::json& j) { ExecuteCommand cmd; + cmd.requestId = j.value("requestId", ""); cmd.traceId = j.value("traceId", ""); cmd.objectRef = j.value("objectRef", ""); cmd.command = j.value("command", ""); @@ -18,6 +21,7 @@ nlohmann::json ExecuteCommand::toJson() const { return nlohmann::json{ {"type", "executeCommand"}, + {"requestId", requestId}, {"traceId", traceId}, {"objectRef", objectRef}, {"command", command}, @@ -45,6 +49,19 @@ bool ExecuteCommand::validate(std::string* error) const } return false; } + if (!params.is_object()) { + if (error) { + *error = "params must be a JSON object"; + } + return false; + } + const auto ref = softbus::core::ObjectRef::fromPath(objectRef); + if (!ref.hasPoint()) { + if (error) { + *error = "objectRef must be a five-layer path (Site/System/Asset/Device/Point)"; + } + return false; + } return true; } diff --git a/src/softbus_bus/CMakeLists.txt b/src/softbus_bus/CMakeLists.txt new file mode 100644 index 0000000..9aeca52 --- /dev/null +++ b/src/softbus_bus/CMakeLists.txt @@ -0,0 +1,8 @@ +softbus_package(softbus_bus) + +add_library(softbus_bus STATIC + src/Bus.cpp +) + +softbus_export_include(softbus_bus "${CMAKE_CURRENT_SOURCE_DIR}/include") +target_link_libraries(softbus_bus PUBLIC softbus_core softbus_api) diff --git a/src/softbus_bus/include/softbus_bus/Bus.h b/src/softbus_bus/include/softbus_bus/Bus.h new file mode 100644 index 0000000..140fe9e --- /dev/null +++ b/src/softbus_bus/include/softbus_bus/Bus.h @@ -0,0 +1,41 @@ +// 消息总线 +// 负责消息的发布和订阅 +#pragma once + +#include + +#include +#include +#include +#include +#include + +namespace softbus::bus { + +using BusHandler = std::function; + +class Bus { +public: + using SubscriptionId = std::uint64_t; + + static Bus& instance(); + + SubscriptionId subscribe(const std::string& topicPattern, BusHandler handler); + void unsubscribe(SubscriptionId id); + void publish(const std::string& topic, const BusMessage& message); + + static bool topicMatches(const std::string& pattern, const std::string& topic); + +private: + struct Subscription { + SubscriptionId id{0}; + std::string pattern; + BusHandler handler; + }; + + std::mutex mutex_; + std::vector subscriptions_; + SubscriptionId nextId_{1}; +}; + +} // namespace softbus::bus diff --git a/src/softbus_bus/include/softbus_bus/BusMessage.h b/src/softbus_bus/include/softbus_bus/BusMessage.h new file mode 100644 index 0000000..2d04c05 --- /dev/null +++ b/src/softbus_bus/include/softbus_bus/BusMessage.h @@ -0,0 +1,31 @@ +#pragma once + +#include +#include +#include + +#include + +namespace softbus::bus { + +enum class BusPayloadType { + DataPoint, + ExecuteCommand, + Ack, + Event, + Heartbeat, + RawPacket +}; + +struct BusMessage { + BusPayloadType type{BusPayloadType::DataPoint}; + std::string topic; + softbus::core::SoftbusDataPoint dataPoint; + softbus::api::ExecuteCommand command; + softbus::core::SoftbusAck ack; + softbus::core::SoftbusEvent event; + softbus::core::RawPacket rawPacket; + std::string deviceId; +}; + +} // namespace softbus::bus diff --git a/src/softbus_bus/src/Bus.cpp b/src/softbus_bus/src/Bus.cpp new file mode 100644 index 0000000..f89da38 --- /dev/null +++ b/src/softbus_bus/src/Bus.cpp @@ -0,0 +1,107 @@ +#include + +#include + +namespace softbus::bus { + +namespace { + +std::vector splitTopic(const std::string& topic) +{ + std::vector parts; + std::string current; + for (char c : topic) { + if (c == '/') { + if (!current.empty()) { + parts.push_back(current); + current.clear(); + } + } else { + current.push_back(c); + } + } + if (!current.empty()) { + parts.push_back(current); + } + return parts; +} + +} // namespace + +Bus& Bus::instance() +{ + static Bus bus; + return bus; +} + +Bus::SubscriptionId Bus::subscribe(const std::string& topicPattern, BusHandler handler) +{ + std::lock_guard lock(mutex_); + const SubscriptionId id = nextId_++; + subscriptions_.push_back(Subscription{id, topicPattern, std::move(handler)}); + return id; +} + +void Bus::unsubscribe(SubscriptionId id) +{ + std::lock_guard lock(mutex_); + subscriptions_.erase( + std::remove_if(subscriptions_.begin(), subscriptions_.end(), + [id](const Subscription& s) { return s.id == id; }), + subscriptions_.end()); +} + +void Bus::publish(const std::string& topic, const BusMessage& message) +{ + std::vector handlers; + { + std::lock_guard lock(mutex_); + for (const auto& sub : subscriptions_) { + if (topicMatches(sub.pattern, topic)) { + handlers.push_back(sub.handler); + } + } + } + BusMessage msg = message; + msg.topic = topic; + for (const auto& handler : handlers) { + handler(msg); + } +} + +bool Bus::topicMatches(const std::string& pattern, const std::string& topic) +{ + if (pattern == topic) { + return true; + } + if (pattern == "#" || pattern == "*") { + return true; + } + + const auto patternParts = splitTopic(pattern); + const auto topicParts = splitTopic(topic); + std::size_t pi = 0; + std::size_t ti = 0; + while (pi < patternParts.size() && ti < topicParts.size()) { + const auto& p = patternParts[pi]; + if (p == "#") { + return true; + } + if (p == "*" || p == "+") { + ++pi; + ++ti; + continue; + } + if (p != topicParts[ti]) { + return false; + } + ++pi; + ++ti; + } + if (pi == patternParts.size() && ti == topicParts.size()) { + return true; + } + return pi < patternParts.size() && patternParts[pi] == "#"; +} + +} // namespace softbus::bus diff --git a/src/softbus_command/CMakeLists.txt b/src/softbus_command/CMakeLists.txt new file mode 100644 index 0000000..e7e4cc4 --- /dev/null +++ b/src/softbus_command/CMakeLists.txt @@ -0,0 +1,9 @@ +softbus_package(softbus_command) + +add_library(softbus_command STATIC + src/CommandContext.cpp + src/CommandService.cpp +) + +softbus_export_include(softbus_command "${CMAKE_CURRENT_SOURCE_DIR}/include") +target_link_libraries(softbus_command PUBLIC softbus_core softbus_api softbus_bus softbus_registry) diff --git a/src/softbus_command/include/softbus_command/CommandContext.h b/src/softbus_command/include/softbus_command/CommandContext.h new file mode 100644 index 0000000..fdacc0a --- /dev/null +++ b/src/softbus_command/include/softbus_command/CommandContext.h @@ -0,0 +1,34 @@ +#pragma once + +#include + +#include +#include + +namespace softbus::command { + +enum class CommandState { + Created, + Validated, + Dispatched, + Delivered, + Executing, + Succeeded, + Failed, + Timeout, + Rejected, + Cancelled +}; + +struct CommandContext { + std::string requestId; + softbus::api::ExecuteCommand command; + CommandState state{CommandState::Created}; + std::string error; + double executedValue{0.0}; + std::chrono::steady_clock::time_point createdAt{std::chrono::steady_clock::now()}; +}; + +std::string commandStateToString(CommandState state); + +} // namespace softbus::command diff --git a/src/softbus_command/include/softbus_command/CommandService.h b/src/softbus_command/include/softbus_command/CommandService.h new file mode 100644 index 0000000..42d362e --- /dev/null +++ b/src/softbus_command/include/softbus_command/CommandService.h @@ -0,0 +1,39 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace softbus::command { + +class CommandService { +public: + explicit CommandService(softbus::registry::PointRegistry* registry, + std::chrono::milliseconds timeout = std::chrono::milliseconds(5000)); + + void attachBus(softbus::bus::Bus* bus); + CommandContext submit(const softbus::api::ExecuteCommand& command, bool waitForAck = true); + std::optional findByRequestId(const std::string& requestId) const; + +private: + void handleAck(const softbus::bus::BusMessage& message); + std::string generateRequestId() const; + + softbus::registry::PointRegistry* registry_{nullptr}; + softbus::bus::Bus* bus_{nullptr}; + softbus::bus::Bus::SubscriptionId ackSub_{0}; + std::chrono::milliseconds timeout_; + mutable std::mutex mutex_; + std::unordered_map contexts_; + std::condition_variable ackCv_; +}; + +} // namespace softbus::command diff --git a/src/softbus_command/src/CommandContext.cpp b/src/softbus_command/src/CommandContext.cpp new file mode 100644 index 0000000..3ca3c19 --- /dev/null +++ b/src/softbus_command/src/CommandContext.cpp @@ -0,0 +1,21 @@ +#include + +namespace softbus::command { + +std::string commandStateToString(CommandState state) +{ + switch (state) { + case CommandState::Validated: return "Validated"; + case CommandState::Dispatched: return "Dispatched"; + case CommandState::Delivered: return "Delivered"; + case CommandState::Executing: return "Executing"; + case CommandState::Succeeded: return "Succeeded"; + case CommandState::Failed: return "Failed"; + case CommandState::Timeout: return "Timeout"; + case CommandState::Rejected: return "Rejected"; + case CommandState::Cancelled: return "Cancelled"; + default: return "Created"; + } +} + +} // namespace softbus::command diff --git a/src/softbus_command/src/CommandService.cpp b/src/softbus_command/src/CommandService.cpp new file mode 100644 index 0000000..7f91919 --- /dev/null +++ b/src/softbus_command/src/CommandService.cpp @@ -0,0 +1,163 @@ +#include + +#include +#include + +namespace softbus::command { + +CommandService::CommandService(softbus::registry::PointRegistry* registry, + std::chrono::milliseconds timeout) + : registry_(registry) + , timeout_(timeout) +{ +} + +void CommandService::attachBus(softbus::bus::Bus* bus) +{ + bus_ = bus; + if (!bus_) { + return; + } + ackSub_ = bus_->subscribe("control/ack/#", [this](const softbus::bus::BusMessage& msg) { + handleAck(msg); + }); +} + +CommandContext CommandService::submit(const softbus::api::ExecuteCommand& command, bool waitForAck) +{ + CommandContext ctx; + ctx.command = command; + ctx.requestId = command.requestId.empty() + ? generateRequestId() + : command.requestId; + ctx.command.requestId = ctx.requestId; + + std::string error; + if (!ctx.command.validate(&error)) { + ctx.state = CommandState::Rejected; + ctx.error = error; + return ctx; + } + + if (!registry_) { + ctx.state = CommandState::Rejected; + ctx.error = "registry not available"; + return ctx; + } + + const auto entry = registry_->findByObjectRef(ctx.command.objectRef); + if (!entry) { + ctx.state = CommandState::Rejected; + ctx.error = "unknown objectRef"; + return ctx; + } + + if (ctx.command.command == "write") { + const double value = ctx.command.params.value("value", 0.0); + if (!registry_->validateWriteValue(*entry, value, &error)) { + ctx.state = CommandState::Rejected; + ctx.error = error; + return ctx; + } + } else if (!entry->writable) { + ctx.state = CommandState::Rejected; + ctx.error = "point is not writable"; + return ctx; + } + + ctx.state = CommandState::Validated; + { + std::lock_guard lock(mutex_); + contexts_[ctx.requestId] = ctx; + } + + if (!bus_) { + ctx.state = CommandState::Failed; + ctx.error = "bus not attached"; + return ctx; + } + + const std::string topic = entry->cmdTopic.empty() + ? registry_->topicRules().formatCommand(entry->deviceId, entry->pointId) + : entry->cmdTopic; + + softbus::bus::BusMessage msg; + msg.type = softbus::bus::BusPayloadType::ExecuteCommand; + msg.command = ctx.command; + bus_->publish(topic, msg); + + { + std::lock_guard lock(mutex_); + contexts_[ctx.requestId].state = CommandState::Dispatched; + } + ctx.state = CommandState::Dispatched; + + SB_LOG_INFO_TRACE("CommandService", ctx.command.traceId, + "command dispatched requestId=", ctx.requestId, " topic=", topic); + + if (!waitForAck) { + return ctx; + } + + std::unique_lock lock(mutex_); + const bool done = ackCv_.wait_for(lock, timeout_, [&]() { + const auto it = contexts_.find(ctx.requestId); + if (it == contexts_.end()) { + return true; + } + return it->second.state == CommandState::Succeeded + || it->second.state == CommandState::Failed + || it->second.state == CommandState::Timeout; + }); + + const auto it = contexts_.find(ctx.requestId); + if (it == contexts_.end()) { + ctx.state = CommandState::Failed; + ctx.error = "context lost"; + return ctx; + } + if (!done && it->second.state == CommandState::Dispatched) { + it->second.state = CommandState::Timeout; + it->second.error = "ack timeout"; + } + return it->second; +} + +std::optional CommandService::findByRequestId(const std::string& requestId) const +{ + std::lock_guard lock(mutex_); + const auto it = contexts_.find(requestId); + if (it == contexts_.end()) { + return std::nullopt; + } + return it->second; +} + +void CommandService::handleAck(const softbus::bus::BusMessage& message) +{ + const auto& ack = message.ack; + if (ack.requestId.empty()) { + return; + } + + std::lock_guard lock(mutex_); + const auto it = contexts_.find(ack.requestId); + if (it == contexts_.end()) { + return; + } + it->second.state = ack.success ? CommandState::Succeeded : CommandState::Failed; + it->second.error = ack.error; + it->second.executedValue = ack.executedValue; + ackCv_.notify_all(); + + SB_LOG_INFO_TRACE("CommandService", ack.traceId, + "ack matched requestId=", ack.requestId, + " state=", commandStateToString(it->second.state)); +} + +std::string CommandService::generateRequestId() const +{ + return "req-" + softbus::core::DeviceIdentity::generateTraceId(); +} + +} // namespace softbus::command diff --git a/src/softbus_core/CMakeLists.txt b/src/softbus_core/CMakeLists.txt index 40ea486..074ad27 100644 --- a/src/softbus_core/CMakeLists.txt +++ b/src/softbus_core/CMakeLists.txt @@ -9,6 +9,7 @@ add_library(softbus_core STATIC src/platform/PlatformSignal.cpp src/identity/DeviceIdentity.cpp src/models/ObjectModel.cpp + src/models/SoftbusTypes.cpp ) softbus_export_include(softbus_core "${CMAKE_CURRENT_SOURCE_DIR}/include") diff --git a/src/softbus_core/include/softbus_core/models/ObjectModel.h b/src/softbus_core/include/softbus_core/models/ObjectModel.h index 1f7933d..1cdb1a6 100644 --- a/src/softbus_core/include/softbus_core/models/ObjectModel.h +++ b/src/softbus_core/include/softbus_core/models/ObjectModel.h @@ -23,12 +23,21 @@ struct ObjectRef { std::string toPath() const; static ObjectRef fromPath(const std::string& path); bool isValid() const; + bool hasPoint() const; }; struct PointDef { std::string name; int registerAddress{0}; std::string domain{"Process"}; + std::string signalType{"AI"}; + std::string physicalChannel; + std::string unit; + double minValue{0.0}; + double maxValue{100.0}; + std::string access{"read"}; + bool writable{false}; + double scale{1.0}; }; struct DeviceDef { diff --git a/src/softbus_core/include/softbus_core/models/SoftbusTypes.h b/src/softbus_core/include/softbus_core/models/SoftbusTypes.h new file mode 100644 index 0000000..33ac899 --- /dev/null +++ b/src/softbus_core/include/softbus_core/models/SoftbusTypes.h @@ -0,0 +1,73 @@ +#pragma once + +#include +#include +#include + +namespace softbus::core { + +enum class SoftbusQuality { + Good, + Bad, + Uncertain, + OutOfRange, + Stale +}; + +enum class SignalType { + AI, + AO, + DI, + DO, + Unknown +}; + +enum class PointAccess { + Read, + Write, + ReadWrite +}; + +struct SoftbusDataPoint { + std::string deviceId; + std::string pointId; + double value{0.0}; + std::string unit; + SoftbusQuality quality{SoftbusQuality::Good}; + std::uint64_t sourceTimestampNs{0}; + std::uint64_t sequence{0}; + std::string objectRef; + + nlohmann::json toJson() const; + static SoftbusDataPoint fromJson(const nlohmann::json& j); +}; + +struct SoftbusAck { + std::string requestId; + std::string traceId; + std::string deviceId; + std::string pointId; + bool success{false}; + std::string error; + double executedValue{0.0}; + + nlohmann::json toJson() const; + static SoftbusAck fromJson(const nlohmann::json& j); +}; + +struct SoftbusEvent { + std::string deviceId; + std::string eventType; + std::string message; + std::uint64_t timestampNs{0}; + + nlohmann::json toJson() const; + static SoftbusEvent fromJson(const nlohmann::json& j); +}; + +std::string qualityToString(SoftbusQuality q); +SignalType signalTypeFromString(const std::string& s); +std::string signalTypeToString(SignalType t); +PointAccess accessFromString(const std::string& s); + +} // namespace softbus::core diff --git a/src/softbus_core/src/models/ObjectModel.cpp b/src/softbus_core/src/models/ObjectModel.cpp index eae4699..8043832 100644 --- a/src/softbus_core/src/models/ObjectModel.cpp +++ b/src/softbus_core/src/models/ObjectModel.cpp @@ -43,6 +43,11 @@ bool ObjectRef::isValid() const return !site.empty() && !system.empty() && !asset.empty() && !device.empty(); } +bool ObjectRef::hasPoint() const +{ + return isValid() && !point.empty(); +} + bool ObjectModelTree::loadFromFile(const std::string& path) { std::ifstream ifs(path); @@ -68,6 +73,22 @@ bool ObjectModelTree::loadFromFile(const std::string& path) point.name = pt.value("name", ""); point.registerAddress = pt.value("register", 0); point.domain = pt.value("domain", "Process"); + point.signalType = pt.value("signalType", "AI"); + point.physicalChannel = pt.value("physicalChannel", ""); + point.unit = pt.value("unit", ""); + point.scale = pt.value("scale", 1.0); + point.access = pt.value("access", "read"); + point.writable = pt.value("writable", false); + if (pt.contains("range") && pt["range"].is_array() && pt["range"].size() >= 2) { + point.minValue = pt["range"][0].get(); + point.maxValue = pt["range"][1].get(); + } else { + point.minValue = pt.value("min", 0.0); + point.maxValue = pt.value("max", 100.0); + } + if (point.access == "read_write" || point.access == "write") { + point.writable = true; + } device.points.push_back(std::move(point)); } } diff --git a/src/softbus_core/src/models/SoftbusTypes.cpp b/src/softbus_core/src/models/SoftbusTypes.cpp new file mode 100644 index 0000000..20dc73c --- /dev/null +++ b/src/softbus_core/src/models/SoftbusTypes.cpp @@ -0,0 +1,128 @@ +#include + +namespace softbus::core { + +nlohmann::json SoftbusDataPoint::toJson() const +{ + return nlohmann::json{ + {"deviceId", deviceId}, + {"pointId", pointId}, + {"value", value}, + {"unit", unit}, + {"quality", qualityToString(quality)}, + {"sourceTimestamp", sourceTimestampNs}, + {"sequence", sequence}, + {"objectRef", objectRef} + }; +} + +SoftbusDataPoint SoftbusDataPoint::fromJson(const nlohmann::json& j) +{ + SoftbusDataPoint dp; + dp.deviceId = j.value("deviceId", ""); + dp.pointId = j.value("pointId", ""); + dp.value = j.value("value", 0.0); + dp.unit = j.value("unit", ""); + dp.sourceTimestampNs = j.value("sourceTimestamp", static_cast(0)); + dp.sequence = j.value("sequence", static_cast(0)); + dp.objectRef = j.value("objectRef", ""); + const auto q = j.value("quality", "GOOD"); + if (q == "BAD") { + dp.quality = SoftbusQuality::Bad; + } else if (q == "OUT_OF_RANGE") { + dp.quality = SoftbusQuality::OutOfRange; + } else if (q == "STALE") { + dp.quality = SoftbusQuality::Stale; + } else if (q == "UNCERTAIN") { + dp.quality = SoftbusQuality::Uncertain; + } + return dp; +} + +nlohmann::json SoftbusAck::toJson() const +{ + return nlohmann::json{ + {"type", "ack"}, + {"requestId", requestId}, + {"traceId", traceId}, + {"deviceId", deviceId}, + {"pointId", pointId}, + {"success", success}, + {"error", error}, + {"executedValue", executedValue} + }; +} + +SoftbusAck SoftbusAck::fromJson(const nlohmann::json& j) +{ + SoftbusAck ack; + ack.requestId = j.value("requestId", ""); + ack.traceId = j.value("traceId", ""); + ack.deviceId = j.value("deviceId", ""); + ack.pointId = j.value("pointId", ""); + ack.success = j.value("success", false); + ack.error = j.value("error", ""); + ack.executedValue = j.value("executedValue", 0.0); + return ack; +} + +nlohmann::json SoftbusEvent::toJson() const +{ + return nlohmann::json{ + {"deviceId", deviceId}, + {"eventType", eventType}, + {"message", message}, + {"timestamp", timestampNs} + }; +} + +SoftbusEvent SoftbusEvent::fromJson(const nlohmann::json& j) +{ + SoftbusEvent ev; + ev.deviceId = j.value("deviceId", ""); + ev.eventType = j.value("eventType", ""); + ev.message = j.value("message", ""); + ev.timestampNs = j.value("timestamp", static_cast(0)); + return ev; +} + +std::string qualityToString(SoftbusQuality q) +{ + switch (q) { + case SoftbusQuality::Bad: return "BAD"; + case SoftbusQuality::Uncertain: return "UNCERTAIN"; + case SoftbusQuality::OutOfRange: return "OUT_OF_RANGE"; + case SoftbusQuality::Stale: return "STALE"; + default: return "GOOD"; + } +} + +SignalType signalTypeFromString(const std::string& s) +{ + if (s == "AI") return SignalType::AI; + if (s == "AO") return SignalType::AO; + if (s == "DI") return SignalType::DI; + if (s == "DO") return SignalType::DO; + return SignalType::Unknown; +} + +std::string signalTypeToString(SignalType t) +{ + switch (t) { + case SignalType::AI: return "AI"; + case SignalType::AO: return "AO"; + case SignalType::DI: return "DI"; + case SignalType::DO: return "DO"; + default: return "Unknown"; + } +} + +PointAccess accessFromString(const std::string& s) +{ + if (s == "read") return PointAccess::Read; + if (s == "write") return PointAccess::Write; + if (s == "read_write") return PointAccess::ReadWrite; + return PointAccess::Read; +} + +} // namespace softbus::core diff --git a/src/softbus_daemon/CMakeLists.txt b/src/softbus_daemon/CMakeLists.txt index f068d17..38524ab 100644 --- a/src/softbus_daemon/CMakeLists.txt +++ b/src/softbus_daemon/CMakeLists.txt @@ -3,10 +3,6 @@ softbus_package(softbus_daemon) add_executable(softbus_daemon src/main.cpp src/CoreService.cpp - src/PipelineEngine.cpp - src/MetadataRegistry.cpp - src/ObjectModelRegistry.cpp - src/CommandDispatcher.cpp src/IpcOptionalDBus.cpp ) @@ -14,6 +10,12 @@ softbus_export_include(softbus_daemon "${CMAKE_CURRENT_SOURCE_DIR}/include") target_link_libraries(softbus_daemon PRIVATE softbus_core + softbus_registry + softbus_bus + softbus_command + softbus_pipeline + softbus_monitor + softbus_storage softbus_transport softbus_api softbus_opcua diff --git a/src/softbus_daemon/src/CoreService.cpp b/src/softbus_daemon/src/CoreService.cpp index db07cb6..56827b4 100644 --- a/src/softbus_daemon/src/CoreService.cpp +++ b/src/softbus_daemon/src/CoreService.cpp @@ -1,14 +1,11 @@ #include -#include -#include - +#include #include #include #include #include #include -#include #include #include @@ -19,6 +16,7 @@ #endif #include +#include #include #include @@ -54,9 +52,27 @@ std::string findPluginInDir(const fs::path& dir, const std::string& pluginName) return {}; } +std::string commandResultJson(const softbus::command::CommandContext& ctx) +{ + nlohmann::json j; + j["success"] = ctx.state == softbus::command::CommandState::Succeeded; + j["requestId"] = ctx.requestId; + j["state"] = softbus::command::commandStateToString(ctx.state); + if (!ctx.error.empty()) { + j["error"] = ctx.error; + } + if (ctx.executedValue != 0.0) { + j["executedValue"] = ctx.executedValue; + } + return j.dump(); +} + } // namespace -CoreService::CoreService() = default; +CoreService::CoreService() + : commandService_(®istry_) +{ +} CoreService::~CoreService() { @@ -84,11 +100,7 @@ std::string CoreService::resolvePluginPath(const std::string& pluginName, exeDir / ".." / "lib", cwd / "build" / "lib", fs::path("build") / "lib", - fs::path("build") / "bin", - cwd / "build" / "bin", - fs::path("lib"), cwd / "lib", - fs::path("bin"), cwd }; @@ -97,69 +109,93 @@ std::string CoreService::resolvePluginPath(const std::string& pluginName, return found; } } - - const auto fallbackNames = pluginFileNames(pluginName); - return fs::absolute(profileDir / "lib" / fallbackNames.front()).string(); + return fs::absolute(profileDir / "lib" / pluginFileNames(pluginName).front()).string(); } +// 主函数启动入口 +// profilePath: 配置文件路径 +// pointTablePath: 点表文件路径 +// return: 是否启动成功 -bool CoreService::start(const std::string& profilePath, const std::string& modelPath) +bool CoreService::start(const std::string& profilePath, const std::string& pointTablePath) { + // 加载配置文件 if (!profile_.loadFromFile(profilePath)) { SB_LOG_ERROR("CoreService", "Failed to load profile: ", profilePath); return false; } - objectRegistry_ = std::make_unique(); - if (!objectRegistry_->loadFromFile(modelPath)) { - SB_LOG_ERROR("CoreService", "Failed to load model: ", modelPath); + const fs::path configDir = fs::path(pointTablePath).parent_path(); + // 加载主题规则文件 + registry_.loadTopicRules((configDir / "topic_rule.json").string()); + if (!registry_.loadPointTable(pointTablePath)) { + SB_LOG_ERROR("CoreService", "Failed to load point table: ", pointTablePath); return false; } - model_ = objectRegistry_->tree(); + model_ = registry_.tree(); const std::string pluginPath = resolvePluginPath(profile_.plugins.modbus, profilePath); if (!modbusPlugin_.load(pluginPath)) { SB_LOG_ERROR("CoreService", "Failed to load modbus plugin: ", pluginPath); return false; } + // 加载点表文件 + auto& bus = softbus::bus::Bus::instance(); + // 将消息总线绑定到命令服务、存储服务和心跳监控服务 + commandService_.attachBus(&bus); + storageService_.attachBus(&bus); + heartbeatMonitor_.attachBus(&bus); + // 创建OPC UA安全配置 softbus::opcua::SecurityConfig security; security.enableTls = profile_.security.tlsEnabled; security.certPath = profile_.security.certPath; security.keyPath = profile_.security.keyPath; + // 创建OPC UA服务引擎 uaEngine_ = std::make_unique(); + // 启动OPC UA服务 if (!uaEngine_->start(profile_.opcua.port, profile_.opcua.applicationName, security)) { return false; } + // 创建OPC UA模型绑定器 uaBinder_ = std::make_unique(uaEngine_->nativeServer()); + // 创建OPC UA方法绑定器 uaMethodBinder_ = std::make_unique(uaEngine_->nativeServer()); + // 创建OPC UA事件桥接器 uaEventBridge_ = std::make_unique(uaEngine_->nativeServer()); + // 将消息总线绑定到OPC UA模型绑定器、OPC UA方法绑定器和OPC UA事件桥接器 + uaBinder_->attachBus(&bus); + uaEventBridge_->attachBus(&bus); + // 创建OPC UA地址空间构建器 softbus::opcua::UaAddressSpaceBuilder builder(uaEngine_->nativeServer()); + // 构建OPC UA地址空间 if (!builder.buildFromModel(model_, uaBinder_.get())) { SB_LOG_ERROR("CoreService", "Failed to build OPC UA address space"); return false; } - pipeline_ = std::make_unique(&modbusPlugin_, uaBinder_.get()); + // 创建数据处理管道 + pipeline_ = std::make_unique( + &modbusPlugin_, uaBinder_.get(), ®istry_); + pipeline_->attachBus(&bus); + // 创建TCP上行监听服务器 uplink_ = std::make_unique( io_, profile_.uplink.listenHost, profile_.uplink.listenPort); - commandDispatcher_ = std::make_unique(uplink_.get()); - - uplink_->setRawPacketHandler([this](const softbus::core::RawPacket& packet) { - SB_LOG_INFO_TRACE("CoreService", packet.traceId, "RawPacket received"); - pipeline_->process(packet); - }); - + // 将消息总线绑定到TCP上行监听服务器 + transportBridge_.attach(&bus, uplink_.get()); + // 绑定OPC UA方法执行命令回调 + // 将OPC UA方法执行命令回调绑定到消息总线 uaMethodBinder_->bindExecuteCommand([this](const softbus::api::ExecuteCommand& cmd) { - return commandDispatcher_->dispatch(cmd); + const auto ctx = commandService_.submit(cmd, true); + return commandResultJson(ctx); }); - + uplink_->start(); ioThread_ = std::thread([this]() { io_.run(); }); running_ = true; - SB_LOG_INFO("CoreService", "Daemon started"); + SB_LOG_INFO("CoreService", "Daemon started (bus+registry+command pipeline)"); return true; } @@ -188,18 +224,24 @@ void CoreService::run(bool selfTest) softbus::core::PlatformSignal::install(nullptr); bool selfTestDone = false; + const auto selfTestStart = std::chrono::steady_clock::now(); while (!softbus::core::PlatformSignal::shouldStop()) { if (uaEngine_) { uaEngine_->iterate(); } - if (selfTest && !selfTestDone && commandDispatcher_) { + heartbeatMonitor_.tick(); + if (selfTest && !selfTestDone + && std::chrono::steady_clock::now() - selfTestStart > std::chrono::seconds(3)) { softbus::api::ExecuteCommand cmd; cmd.traceId = softbus::core::DeviceIdentity::generateTraceId(); - cmd.objectRef = "DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature"; + cmd.objectRef = "DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening"; cmd.command = "write"; - cmd.params = {{"value", 42.5}}; - commandDispatcher_->dispatch(cmd); + cmd.params = {{"value", 60.0}}; + const auto ctx = commandService_.submit(cmd, true); + SB_LOG_INFO("CoreService", "self-test command state=", + softbus::command::commandStateToString(ctx.state), + " executedValue=", ctx.executedValue); selfTestDone = true; } std::this_thread::sleep_for(std::chrono::milliseconds(10)); diff --git a/src/softbus_daemon/src/main.cpp b/src/softbus_daemon/src/main.cpp index 57d6d0d..988cfd7 100644 --- a/src/softbus_daemon/src/main.cpp +++ b/src/softbus_daemon/src/main.cpp @@ -6,22 +6,22 @@ int main(int argc, char* argv[]) { std::string profilePath = "config/daemon_profile.json"; - std::string modelPath = "config/softbus_model.json"; + std::string pointTablePath = "config/point_table.json"; bool selfTest = false; for (int i = 1; i < argc; ++i) { const std::string arg = argv[i]; if (arg == "--config" && i + 1 < argc) { profilePath = argv[++i]; - } else if (arg == "--model" && i + 1 < argc) { - modelPath = argv[++i]; + } else if ((arg == "--point-table" || arg == "--model") && i + 1 < argc) { + pointTablePath = argv[++i]; } else if (arg == "--self-test") { selfTest = true; } } softbus::daemon::CoreService service; - if (!service.start(profilePath, modelPath)) { + if (!service.start(profilePath, pointTablePath)) { std::cerr << "Failed to start softbus_daemon" << std::endl; return 1; } diff --git a/src/softbus_edge/CMakeLists.txt b/src/softbus_edge/CMakeLists.txt index d4b8ac7..d64b7f2 100644 --- a/src/softbus_edge/CMakeLists.txt +++ b/src/softbus_edge/CMakeLists.txt @@ -6,9 +6,18 @@ add_executable(softbus_edge src/UplinkClient.cpp src/DeviceFactory.cpp src/DeviceDiscovery.cpp + src/EdgeConfig.cpp + src/ShadowChannelStore.cpp + src/CommandExecutor.cpp ) softbus_export_include(softbus_edge "${CMAKE_CURRENT_SOURCE_DIR}/include") -target_link_libraries(softbus_edge PRIVATE softbus_core softbus_transport softbus_api asio) +target_link_libraries(softbus_edge PRIVATE + softbus_core + softbus_registry + softbus_transport + softbus_api + asio +) softbus_apply_platform_libs(softbus_edge) diff --git a/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h b/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h new file mode 100644 index 0000000..6a15fcc --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/config/EdgeConfig.h @@ -0,0 +1,18 @@ +#pragma once + +#include +#include + +namespace softbus::edge { + +struct EdgeConfig { + std::string deviceId{"EdgeDevice_01"}; + std::string daemonHost{"127.0.0.1"}; + uint16_t daemonPort{9000}; + int heartbeatPeriodMs{1000}; + std::string pointTablePath{"config/point_table.json"}; + + bool loadFromFile(const std::string& path); +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h b/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h new file mode 100644 index 0000000..0504ef1 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/egress/CommandExecutor.h @@ -0,0 +1,26 @@ +#pragma once + +#include +#include +#include +#include + +namespace softbus::edge { + +class CommandExecutor { +public: + CommandExecutor(softbus::registry::PointRegistry* registry, + ShadowChannelStore* store, + UplinkClient* uplink, + std::string deviceId); + + softbus::core::SoftbusAck execute(const softbus::api::ExecuteCommand& command); + +private: + softbus::registry::PointRegistry* registry_{nullptr}; + ShadowChannelStore* store_{nullptr}; + UplinkClient* uplink_{nullptr}; + std::string deviceId_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h b/src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h new file mode 100644 index 0000000..96c53f1 --- /dev/null +++ b/src/softbus_edge/include/softbus_edge/egress/ShadowChannelStore.h @@ -0,0 +1,21 @@ +#pragma once + +#include + +#include +#include +#include + +namespace softbus::edge { + +class ShadowChannelStore { +public: + bool write(const softbus::registry::PointTableEntry& entry, double engineeringValue); + std::optional read(const std::string& physicalChannel) const; + int rawRegisterValue(const softbus::registry::PointTableEntry& entry, double engineeringValue) const; + +private: + std::unordered_map channels_; +}; + +} // namespace softbus::edge diff --git a/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h b/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h index 47f6dc5..593c2cb 100644 --- a/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h +++ b/src/softbus_edge/include/softbus_edge/egress/UplinkClient.h @@ -2,6 +2,7 @@ #include #include +#include #include #include @@ -18,6 +19,8 @@ public: bool connect(); void disconnect(); bool sendRawPacket(const softbus::core::RawPacket& packet); + bool sendAck(const softbus::core::SoftbusAck& ack); + bool sendHeartbeat(const std::string& deviceId); void setCommandHandler(std::function handler); void poll(); diff --git a/src/softbus_edge/src/CommandExecutor.cpp b/src/softbus_edge/src/CommandExecutor.cpp new file mode 100644 index 0000000..703ecdd --- /dev/null +++ b/src/softbus_edge/src/CommandExecutor.cpp @@ -0,0 +1,69 @@ +#include + +#include + +namespace softbus::edge { + +CommandExecutor::CommandExecutor(softbus::registry::PointRegistry* registry, + ShadowChannelStore* store, + UplinkClient* uplink, + std::string deviceId) + : registry_(registry) + , store_(store) + , uplink_(uplink) + , deviceId_(std::move(deviceId)) +{ +} + +softbus::core::SoftbusAck CommandExecutor::execute(const softbus::api::ExecuteCommand& command) +{ + softbus::core::SoftbusAck ack; + ack.requestId = command.requestId; + ack.traceId = command.traceId; + + if (!registry_ || !store_) { + ack.success = false; + ack.error = "executor not initialized"; + return ack; + } + + const auto entry = registry_->findByObjectRef(command.objectRef); + if (!entry) { + ack.success = false; + ack.error = "unknown objectRef on edge"; + return ack; + } + + ack.deviceId = entry->deviceId; + ack.pointId = entry->pointId; + + if (command.command != "write") { + ack.success = false; + ack.error = "unsupported command"; + return ack; + } + + const double value = command.params.value("value", 0.0); + std::string error; + if (!registry_->validateWriteValue(*entry, value, &error)) { + ack.success = false; + ack.error = error; + return ack; + } + + store_->write(*entry, value); + ack.executedValue = value; + ack.success = true; + + const int raw = store_->rawRegisterValue(*entry, value); + SB_LOG_INFO_TRACE("CommandExecutor", command.traceId, + "write ", entry->physicalChannel, " value=", value, + entry->unit, " raw=", raw, " (mock AO)"); + + if (uplink_) { + uplink_->sendAck(ack); + } + return ack; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/EdgeConfig.cpp b/src/softbus_edge/src/EdgeConfig.cpp new file mode 100644 index 0000000..f26eed1 --- /dev/null +++ b/src/softbus_edge/src/EdgeConfig.cpp @@ -0,0 +1,28 @@ +#include + +#include +#include + +namespace softbus::edge { + +bool EdgeConfig::loadFromFile(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + if (!j.contains("edge")) { + return false; + } + const auto& edge = j["edge"]; + deviceId = edge.value("deviceId", deviceId); + daemonHost = edge.value("daemonHost", daemonHost); + daemonPort = static_cast(edge.value("daemonPort", static_cast(daemonPort))); + heartbeatPeriodMs = edge.value("heartbeatPeriodMs", heartbeatPeriodMs); + pointTablePath = edge.value("pointTable", pointTablePath); + return true; +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/ShadowChannelStore.cpp b/src/softbus_edge/src/ShadowChannelStore.cpp new file mode 100644 index 0000000..5554cf9 --- /dev/null +++ b/src/softbus_edge/src/ShadowChannelStore.cpp @@ -0,0 +1,32 @@ +#include + +#include + +namespace softbus::edge { + +bool ShadowChannelStore::write(const softbus::registry::PointTableEntry& entry, + double engineeringValue) +{ + channels_[entry.physicalChannel] = engineeringValue; + return true; +} + +std::optional ShadowChannelStore::read(const std::string& physicalChannel) const +{ + const auto it = channels_.find(physicalChannel); + if (it == channels_.end()) { + return std::nullopt; + } + return it->second; +} + +int ShadowChannelStore::rawRegisterValue(const softbus::registry::PointTableEntry& entry, + double engineeringValue) const +{ + if (entry.scale <= 0.0) { + return static_cast(std::lround(engineeringValue)); + } + return static_cast(std::lround(engineeringValue / entry.scale)); +} + +} // namespace softbus::edge diff --git a/src/softbus_edge/src/UplinkClient.cpp b/src/softbus_edge/src/UplinkClient.cpp index a3a2d7b..30996cc 100644 --- a/src/softbus_edge/src/UplinkClient.cpp +++ b/src/softbus_edge/src/UplinkClient.cpp @@ -27,6 +27,16 @@ bool UplinkClient::sendRawPacket(const softbus::core::RawPacket& packet) return client_->sendRawPacket(packet); } +bool UplinkClient::sendAck(const softbus::core::SoftbusAck& ack) +{ + return client_->sendAck(ack); +} + +bool UplinkClient::sendHeartbeat(const std::string& deviceId) +{ + return client_->sendHeartbeat(deviceId); +} + void UplinkClient::setCommandHandler(std::function handler) { client_->setCommandHandler(std::move(handler)); diff --git a/src/softbus_edge/src/main.cpp b/src/softbus_edge/src/main.cpp index 693e2b7..c58ac3f 100644 --- a/src/softbus_edge/src/main.cpp +++ b/src/softbus_edge/src/main.cpp @@ -1,11 +1,17 @@ +#include +#include +#include #include -#include #include +#include #include +#include #include - +#include +#include #include +#include #include #include #include @@ -13,9 +19,8 @@ namespace { struct EdgeOptions { - std::string daemonHost{"127.0.0.1"}; - uint16_t daemonPort{9000}; - std::string objectRef{"DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature"}; + std::string configPath{"config/edge.json"}; + std::string objectRef{"DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature"}; }; EdgeOptions parseArgs(int argc, char* argv[]) @@ -23,15 +28,8 @@ EdgeOptions parseArgs(int argc, char* argv[]) EdgeOptions options; for (int i = 1; i < argc; ++i) { const std::string arg = argv[i]; - if (arg == "--daemon" && i + 1 < argc) { - const std::string endpoint = argv[++i]; - const auto pos = endpoint.find(':'); - if (pos != std::string::npos) { - options.daemonHost = endpoint.substr(0, pos); - options.daemonPort = static_cast(std::stoi(endpoint.substr(pos + 1))); - } else { - options.daemonHost = endpoint; - } + if (arg == "--config" && i + 1 < argc) { + options.configPath = argv[++i]; } else if (arg == "--object-ref" && i + 1 < argc) { options.objectRef = argv[++i]; } @@ -45,36 +43,83 @@ int main(int argc, char* argv[]) { const EdgeOptions options = parseArgs(argc, argv); - softbus::edge::DeviceFactory factory; - const std::string runtimeDeviceId = factory.createRuntimeDeviceId(); - SB_LOG_INFO("Edge", "runtimeDeviceId=", runtimeDeviceId, " (local only, not persisted)"); + softbus::edge::EdgeConfig config; + if (!config.loadFromFile(options.configPath)) { + std::cerr << "Failed to load edge config: " << options.configPath << std::endl; + return 1; + } - softbus::edge::UplinkClient uplink(options.daemonHost, options.daemonPort); - uplink.setCommandHandler([](const softbus::api::ExecuteCommand& cmd) { - SB_LOG_INFO_TRACE("Edge", cmd.traceId, - "executeCommand received: ", cmd.toJson().dump()); + softbus::registry::PointRegistry registry; + const auto configDir = std::filesystem::path(config.pointTablePath).parent_path(); + registry.loadTopicRules((configDir / "topic_rule.json").string()); + if (!registry.loadPointTable(config.pointTablePath)) { + std::cerr << "Failed to load point table: " << config.pointTablePath << std::endl; + return 1; + } + + softbus::edge::ShadowChannelStore shadowStore; + softbus::edge::UplinkClient uplink(config.daemonHost, config.daemonPort); + softbus::edge::CommandExecutor executor(®istry, &shadowStore, &uplink, config.deviceId); + uplink.setCommandHandler([&executor](const softbus::api::ExecuteCommand& cmd) { + const auto ack = executor.execute(cmd); + if (ack.success) { + SB_LOG_INFO_TRACE("Edge", cmd.traceId, + "command executed point=", ack.pointId, " value=", ack.executedValue); + } else { + SB_LOG_WARN_TRACE("Edge", cmd.traceId, "command failed: ", ack.error); + } }); if (!uplink.connect()) { std::cerr << "Failed to connect to daemon at " - << options.daemonHost << ":" << options.daemonPort << std::endl; + << config.daemonHost << ":" << config.daemonPort << std::endl; return 1; } - softbus::edge::MockModbusIngress ingress(options.objectRef, runtimeDeviceId); + softbus::edge::MockModbusIngress ingress(options.objectRef, config.deviceId); ingress.open(); softbus::core::PlatformSignal::install([]() { SB_LOG_INFO("Edge", "Shutdown requested"); }); + auto lastHeartbeat = std::chrono::steady_clock::now(); + while (!softbus::core::PlatformSignal::shouldStop()) { uplink.poll(); + + const auto now = std::chrono::steady_clock::now(); + if (now - lastHeartbeat >= std::chrono::milliseconds(config.heartbeatPeriodMs)) { + uplink.sendHeartbeat(config.deviceId); + lastHeartbeat = now; + } + for (const auto& packet : ingress.poll()) { SB_LOG_INFO_TRACE("Edge", packet.traceId, "Uplink RawPacket objectRef=", packet.sourceId); uplink.sendRawPacket(packet); } + + for (const auto& entry : registry.entries()) { + const auto value = shadowStore.read(entry.physicalChannel); + if (!value) { + continue; + } + softbus::core::RawPacket feedback; + feedback.traceId = softbus::core::DeviceIdentity::generateTraceId(); + feedback.timestampNs = softbus::core::HwClock::nowNs(); + feedback.sourceId = entry.objectRef; + feedback.protocolTag = "shadow"; + const int raw = shadowStore.rawRegisterValue(entry, *value); + feedback.payload = { + static_cast(entry.registerAddress >> 8), + static_cast(entry.registerAddress & 0xFF), + static_cast(raw >> 8), + static_cast(raw & 0xFF) + }; + uplink.sendRawPacket(feedback); + } + std::this_thread::sleep_for(std::chrono::seconds(1)); } diff --git a/src/softbus_monitor/CMakeLists.txt b/src/softbus_monitor/CMakeLists.txt new file mode 100644 index 0000000..7934f38 --- /dev/null +++ b/src/softbus_monitor/CMakeLists.txt @@ -0,0 +1,8 @@ +softbus_package(softbus_monitor) + +add_library(softbus_monitor STATIC + src/HeartbeatMonitor.cpp +) + +softbus_export_include(softbus_monitor "${CMAKE_CURRENT_SOURCE_DIR}/include") +target_link_libraries(softbus_monitor PUBLIC softbus_core softbus_bus) diff --git a/src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h b/src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h new file mode 100644 index 0000000..9f4c4f9 --- /dev/null +++ b/src/softbus_monitor/include/softbus_monitor/HeartbeatMonitor.h @@ -0,0 +1,30 @@ +#pragma once + +#include + +#include +#include +#include +#include + +namespace softbus::monitor { + +class HeartbeatMonitor { +public: + explicit HeartbeatMonitor(std::chrono::milliseconds timeout = std::chrono::seconds(5)); + + void attachBus(softbus::bus::Bus* bus); + void tick(); + +private: + void onHeartbeat(const softbus::bus::BusMessage& message); + void publishOfflineEvent(const std::string& deviceId); + + softbus::bus::Bus* bus_{nullptr}; + softbus::bus::Bus::SubscriptionId hbSub_{0}; + std::chrono::milliseconds timeout_; + std::mutex mutex_; + std::unordered_map lastSeen_; +}; + +} // namespace softbus::monitor diff --git a/src/softbus_monitor/src/HeartbeatMonitor.cpp b/src/softbus_monitor/src/HeartbeatMonitor.cpp new file mode 100644 index 0000000..05467a9 --- /dev/null +++ b/src/softbus_monitor/src/HeartbeatMonitor.cpp @@ -0,0 +1,70 @@ +#include + +#include +#include + +#include + +namespace softbus::monitor { + +HeartbeatMonitor::HeartbeatMonitor(std::chrono::milliseconds timeout) + : timeout_(timeout) +{ +} + +void HeartbeatMonitor::attachBus(softbus::bus::Bus* bus) +{ + bus_ = bus; + if (!bus_) { + return; + } + hbSub_ = bus_->subscribe("system/heartbeat/#", [this](const softbus::bus::BusMessage& msg) { + onHeartbeat(msg); + }); +} + +void HeartbeatMonitor::onHeartbeat(const softbus::bus::BusMessage& message) +{ + if (message.deviceId.empty()) { + return; + } + std::lock_guard lock(mutex_); + lastSeen_[message.deviceId] = std::chrono::steady_clock::now(); +} + +void HeartbeatMonitor::tick() +{ + const auto now = std::chrono::steady_clock::now(); + std::vector offline; + { + std::lock_guard lock(mutex_); + for (const auto& [deviceId, last] : lastSeen_) { + if (now - last > timeout_) { + offline.push_back(deviceId); + } + } + } + for (const auto& deviceId : offline) { + publishOfflineEvent(deviceId); + } +} + +void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId) +{ + if (!bus_) { + return; + } + softbus::core::SoftbusEvent ev; + ev.deviceId = deviceId; + ev.eventType = "edge_offline"; + ev.message = "heartbeat timeout"; + ev.timestampNs = softbus::core::HwClock::nowNs(); + + softbus::bus::BusMessage msg; + msg.type = softbus::bus::BusPayloadType::Event; + msg.event = ev; + bus_->publish("event/" + deviceId + "/edge_offline", msg); + SB_LOG_WARN("HeartbeatMonitor", "edge offline: ", deviceId); +} + +} // namespace softbus::monitor diff --git a/src/softbus_opcua/CMakeLists.txt b/src/softbus_opcua/CMakeLists.txt index 38f89c7..c29c5da 100644 --- a/src/softbus_opcua/CMakeLists.txt +++ b/src/softbus_opcua/CMakeLists.txt @@ -10,6 +10,6 @@ add_library(softbus_opcua STATIC softbus_export_include(softbus_opcua "${CMAKE_CURRENT_SOURCE_DIR}/include") -target_link_libraries(softbus_opcua PUBLIC softbus_core softbus_api) +target_link_libraries(softbus_opcua PUBLIC softbus_core softbus_api softbus_bus) softbus_apply_open62541(softbus_opcua) softbus_apply_platform_libs(softbus_opcua) diff --git a/src/softbus_opcua/include/softbus_opcua/UaEventBridge.h b/src/softbus_opcua/include/softbus_opcua/UaEventBridge.h index 7133a9c..62a71ad 100644 --- a/src/softbus_opcua/include/softbus_opcua/UaEventBridge.h +++ b/src/softbus_opcua/include/softbus_opcua/UaEventBridge.h @@ -1,5 +1,7 @@ #pragma once +#include + #include struct UA_Server; @@ -14,9 +16,14 @@ public: void emitDeviceOnline(const std::string& stableDeviceKey); void emitDeviceOffline(const std::string& stableDeviceKey); void emitParseError(const std::string& objectRef, const std::string& detail); + void attachBus(softbus::bus::Bus* bus); private: + void onBusMessage(const softbus::bus::BusMessage& message); + UA_Server* server_{nullptr}; + softbus::bus::Bus* bus_{nullptr}; + softbus::bus::Bus::SubscriptionId eventSub_{0}; }; } // namespace softbus::opcua diff --git a/src/softbus_opcua/include/softbus_opcua/UaMethodBinder.h b/src/softbus_opcua/include/softbus_opcua/UaMethodBinder.h index dc7daa9..d6e23ae 100644 --- a/src/softbus_opcua/include/softbus_opcua/UaMethodBinder.h +++ b/src/softbus_opcua/include/softbus_opcua/UaMethodBinder.h @@ -9,14 +9,14 @@ struct UA_Server; namespace softbus::opcua { -using ExecuteCommandHandler = std::function; +using ExecuteCommandHandler = std::function; class UaMethodBinder { public: explicit UaMethodBinder(UA_Server* server); bool bindExecuteCommand(ExecuteCommandHandler handler); - bool dispatchCommand(const softbus::api::ExecuteCommand& command); + std::string dispatchCommand(const softbus::api::ExecuteCommand& command); private: UA_Server* server_{nullptr}; diff --git a/src/softbus_opcua/include/softbus_opcua/UaModelBinder.h b/src/softbus_opcua/include/softbus_opcua/UaModelBinder.h index e7f712a..ae36c02 100644 --- a/src/softbus_opcua/include/softbus_opcua/UaModelBinder.h +++ b/src/softbus_opcua/include/softbus_opcua/UaModelBinder.h @@ -1,10 +1,13 @@ #pragma once +#include #include #include #include +struct UA_Server; + #if SOFTBUS_HAS_OPCUA #include #else @@ -18,10 +21,16 @@ public: explicit UaModelBinder(UA_Server* server); void registerPointNode(const std::string& objectRef, const UA_NodeId& nodeId); + void attachBus(softbus::bus::Bus* bus); bool bind(const softbus::core::DOMMessage& message); + bool bindValue(const std::string& objectRef, double value, const std::string& traceId = ""); private: + void onBusMessage(const softbus::bus::BusMessage& message); + UA_Server* server_{nullptr}; + softbus::bus::Bus* bus_{nullptr}; + softbus::bus::Bus::SubscriptionId dataSub_{0}; std::map pointNodes_; }; diff --git a/src/softbus_opcua/src/UaEventBridge.cpp b/src/softbus_opcua/src/UaEventBridge.cpp index 7b76d35..bc60d9d 100644 --- a/src/softbus_opcua/src/UaEventBridge.cpp +++ b/src/softbus_opcua/src/UaEventBridge.cpp @@ -9,6 +9,25 @@ UaEventBridge::UaEventBridge(UA_Server* server) { } +void UaEventBridge::attachBus(softbus::bus::Bus* bus) +{ + bus_ = bus; + if (!bus_) { + return; + } + eventSub_ = bus_->subscribe("event/#", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); +} + +void UaEventBridge::onBusMessage(const softbus::bus::BusMessage& message) +{ + if (message.type != softbus::bus::BusPayloadType::Event) { + return; + } + emitDeviceOffline(message.event.deviceId + ": " + message.event.message); +} + void UaEventBridge::emitHeartbeat(const std::string& objectRef) { SB_LOG_DEBUG("UaEventBridge", "Heartbeat stub for ", objectRef); diff --git a/src/softbus_opcua/src/UaMethodBinder.cpp b/src/softbus_opcua/src/UaMethodBinder.cpp index 6a02ea0..e572748 100644 --- a/src/softbus_opcua/src/UaMethodBinder.cpp +++ b/src/softbus_opcua/src/UaMethodBinder.cpp @@ -60,8 +60,7 @@ UA_StatusCode executeCommandMethod(UA_Server* server, return UA_STATUSCODE_GOOD; } - const bool ok = g_methodBinder->dispatchCommand(cmd); - const std::string result = ok ? R"({"success":true})" : R"({"success":false})"; + const std::string result = g_methodBinder->dispatchCommand(cmd); UA_String out = UA_STRING_ALLOC(result.c_str()); UA_Variant_setScalarCopy(output, &out, &UA_TYPES[UA_TYPES_STRING]); UA_String_clear(&out); @@ -75,10 +74,10 @@ UaMethodBinder::UaMethodBinder(UA_Server* server) { } -bool UaMethodBinder::dispatchCommand(const softbus::api::ExecuteCommand& command) +std::string UaMethodBinder::dispatchCommand(const softbus::api::ExecuteCommand& command) { if (!handler_) { - return false; + return R"({"success":false,"error":"handler not bound"})"; } SB_LOG_INFO_TRACE("UaMethodBinder", command.traceId, "ExecuteCommand invoked via OPC UA method"); diff --git a/src/softbus_opcua/src/UaModelBinder.cpp b/src/softbus_opcua/src/UaModelBinder.cpp index 23f024c..73bfd58 100644 --- a/src/softbus_opcua/src/UaModelBinder.cpp +++ b/src/softbus_opcua/src/UaModelBinder.cpp @@ -13,6 +13,25 @@ UaModelBinder::UaModelBinder(UA_Server* server) { } +void UaModelBinder::attachBus(softbus::bus::Bus* bus) +{ + bus_ = bus; + if (!bus_) { + return; + } + dataSub_ = bus_->subscribe("data/#", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); +} + +void UaModelBinder::onBusMessage(const softbus::bus::BusMessage& message) +{ + if (message.type != softbus::bus::BusPayloadType::DataPoint) { + return; + } + bindValue(message.dataPoint.objectRef, message.dataPoint.value); +} + void UaModelBinder::registerPointNode(const std::string& objectRef, const UA_NodeId& nodeId) { #if SOFTBUS_HAS_OPCUA @@ -27,43 +46,47 @@ void UaModelBinder::registerPointNode(const std::string& objectRef, const UA_Nod bool UaModelBinder::bind(const softbus::core::DOMMessage& message) { -#if !SOFTBUS_HAS_OPCUA - (void)message; - return false; -#else - if (!server_) { - return false; - } - const std::string key = message.id.empty() ? message.sourceId : message.id; - const auto it = pointNodes_.find(key); - if (it == pointNodes_.end()) { - SB_LOG_WARN_TRACE("UaModelBinder", message.traceId, - "No OPC UA node registered for ", key); - return false; - } - double numericValue = 0.0; if (message.value.contains("value")) { numericValue = message.value["value"].get(); } else if (message.value.contains("registerValue")) { numericValue = message.value["registerValue"].get(); } + return bindValue(key, numericValue, message.traceId); +} + +bool UaModelBinder::bindValue(const std::string& objectRef, double value, const std::string& traceId) +{ +#if !SOFTBUS_HAS_OPCUA + (void)objectRef; + (void)value; + (void)traceId; + return false; +#else + if (!server_) { + return false; + } + + const auto it = pointNodes_.find(objectRef); + if (it == pointNodes_.end()) { + SB_LOG_WARN_TRACE("UaModelBinder", traceId, "No OPC UA node registered for ", objectRef); + return false; + } UA_Variant variant; UA_Variant_init(&variant); - UA_Variant_setScalarCopy(&variant, &numericValue, &UA_TYPES[UA_TYPES_DOUBLE]); + UA_Variant_setScalarCopy(&variant, &value, &UA_TYPES[UA_TYPES_DOUBLE]); const UA_StatusCode status = UA_Server_writeValue(server_, it->second, variant); UA_Variant_clear(&variant); if (status == UA_STATUSCODE_GOOD) { - SB_LOG_INFO_TRACE("UaModelBinder", message.traceId, - "Bound DOMMessage to OPC UA node ", key, " value=", numericValue); + SB_LOG_INFO_TRACE("UaModelBinder", traceId, + "Bound value to OPC UA node ", objectRef, " value=", value); return true; } - SB_LOG_ERROR_TRACE("UaModelBinder", message.traceId, - "Failed to write OPC UA node ", key); + SB_LOG_ERROR_TRACE("UaModelBinder", traceId, "Failed to write OPC UA node ", objectRef); return false; #endif } diff --git a/src/softbus_pipeline/CMakeLists.txt b/src/softbus_pipeline/CMakeLists.txt new file mode 100644 index 0000000..a0c08af --- /dev/null +++ b/src/softbus_pipeline/CMakeLists.txt @@ -0,0 +1,10 @@ +softbus_package(softbus_pipeline) + +add_library(softbus_pipeline STATIC + src/PipelineEngine.cpp +) + +softbus_export_include(softbus_pipeline "${CMAKE_CURRENT_SOURCE_DIR}/include") +target_link_libraries(softbus_pipeline PUBLIC + softbus_core softbus_bus softbus_registry softbus_opcua +) diff --git a/src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h b/src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h new file mode 100644 index 0000000..ed092dc --- /dev/null +++ b/src/softbus_pipeline/include/softbus_pipeline/PipelineEngine.h @@ -0,0 +1,31 @@ +#pragma once + +#include +#include +#include +#include + +namespace softbus::pipeline { + +class PipelineEngine { +public: + PipelineEngine(softbus::core::PluginLoader* plugin, + softbus::opcua::UaModelBinder* binder, + softbus::registry::PointRegistry* registry); + + void attachBus(softbus::bus::Bus* bus); + void process(const softbus::core::RawPacket& packet); + +private: + void onBusMessage(const softbus::bus::BusMessage& message); + void publishDataPoint(const softbus::core::DOMMessage& message, + const softbus::registry::PointTableEntry& entry); + + softbus::core::PluginLoader* plugin_{nullptr}; + softbus::opcua::UaModelBinder* binder_{nullptr}; + softbus::registry::PointRegistry* registry_{nullptr}; + softbus::bus::Bus* bus_{nullptr}; + softbus::bus::Bus::SubscriptionId rawSub_{0}; +}; + +} // namespace softbus::pipeline diff --git a/src/softbus_pipeline/src/PipelineEngine.cpp b/src/softbus_pipeline/src/PipelineEngine.cpp new file mode 100644 index 0000000..fb0a04a --- /dev/null +++ b/src/softbus_pipeline/src/PipelineEngine.cpp @@ -0,0 +1,108 @@ +#include + +#include +#include + +#include + +namespace softbus::pipeline { + +PipelineEngine::PipelineEngine(softbus::core::PluginLoader* plugin, + softbus::opcua::UaModelBinder* binder, + softbus::registry::PointRegistry* registry) + : plugin_(plugin) + , binder_(binder) + , registry_(registry) +{ +} + +void PipelineEngine::attachBus(softbus::bus::Bus* bus) +{ + bus_ = bus; + if (!bus_) { + return; + } + rawSub_ = bus_->subscribe("edge/raw/#", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); +} + +void PipelineEngine::onBusMessage(const softbus::bus::BusMessage& message) +{ + if (message.type == softbus::bus::BusPayloadType::RawPacket) { + process(message.rawPacket); + } +} + +void PipelineEngine::process(const softbus::core::RawPacket& packet) +{ + if (!plugin_ || !plugin_->parser() || !binder_) { + return; + } + + const uint8_t* data = packet.payload.data(); + std::size_t size = packet.payload.size(); + std::vector frame = packet.payload; + + if (plugin_->framer()) { + const auto framed = plugin_->framer()->extract(data, size); + if (framed.success) { + frame = framed.frame; + data = frame.data(); + size = frame.size(); + } + } + + auto parsed = plugin_->parser()->parseFrame(data, size, packet.sourceId); + if (!parsed.success) { + SB_LOG_WARN_TRACE("PipelineEngine", packet.traceId, "Failed to parse RawPacket"); + return; + } + + parsed.message.traceId = packet.traceId; + parsed.message.timestampNs = packet.timestampNs; + + if (registry_) { + const auto entry = registry_->findByObjectRef(parsed.message.id); + if (entry) { + publishDataPoint(parsed.message, *entry); + } + } + + binder_->bind(parsed.message); + SB_LOG_INFO_TRACE("PipelineEngine", packet.traceId, + "RawPacket converted to DOMMessage id=", parsed.message.id); +} + +void PipelineEngine::publishDataPoint(const softbus::core::DOMMessage& message, + const softbus::registry::PointTableEntry& entry) +{ + if (!bus_) { + return; + } + + double value = 0.0; + if (message.value.contains("value")) { + value = message.value["value"].get(); + } else if (message.value.contains("registerValue")) { + value = message.value["registerValue"].get(); + } + + softbus::core::SoftbusDataPoint dp; + dp.deviceId = entry.deviceId; + dp.pointId = entry.pointId; + dp.value = value; + dp.unit = entry.unit; + dp.objectRef = entry.objectRef; + dp.sourceTimestampNs = message.timestampNs; + + softbus::bus::BusMessage busMsg; + busMsg.type = softbus::bus::BusPayloadType::DataPoint; + busMsg.dataPoint = dp; + const std::string topic = entry.dataTopic.empty() + ? registry_->topicRules().formatData(entry.deviceId, entry.pointId) + : entry.dataTopic; + bus_->publish(topic, busMsg); +} + +} // namespace softbus::pipeline diff --git a/src/softbus_registry/CMakeLists.txt b/src/softbus_registry/CMakeLists.txt new file mode 100644 index 0000000..51720f3 --- /dev/null +++ b/src/softbus_registry/CMakeLists.txt @@ -0,0 +1,9 @@ +softbus_package(softbus_registry) + +add_library(softbus_registry STATIC + src/PointRegistry.cpp + src/PointTableEntry.cpp +) + +softbus_export_include(softbus_registry "${CMAKE_CURRENT_SOURCE_DIR}/include") +target_link_libraries(softbus_registry PUBLIC softbus_core nlohmann_json::nlohmann_json) diff --git a/src/softbus_registry/include/softbus_registry/PointRegistry.h b/src/softbus_registry/include/softbus_registry/PointRegistry.h new file mode 100644 index 0000000..ec0bfd2 --- /dev/null +++ b/src/softbus_registry/include/softbus_registry/PointRegistry.h @@ -0,0 +1,42 @@ +#pragma once + +#include +#include + +#include +#include +#include +#include + +namespace softbus::registry { + +class PointRegistry { +public: + bool loadPointTable(const std::string& path); + bool loadTopicRules(const std::string& path); + + const softbus::core::ObjectModelTree& tree() const { return tree_; } + const TopicRules& topicRules() const { return topicRules_; } + const std::vector& entries() const { return entries_; } + + std::optional findByObjectRef(const std::string& objectRef) const; + std::optional findByTopic(const std::string& topic) const; + std::optional findByOpcuaNodeId(const std::string& nodeId) const; + std::optional findByDevicePoint(const std::string& deviceId, + const std::string& pointId) const; + std::optional findPointByRegister(int registerAddress) const; + bool validateWriteValue(const PointTableEntry& entry, double value, std::string* error) const; + +private: + void indexEntry(const PointTableEntry& entry); + + softbus::core::ObjectModelTree tree_; + TopicRules topicRules_; + std::vector entries_; + std::unordered_map byObjectRef_; + std::unordered_map byTopic_; + std::unordered_map byOpcuaNodeId_; + std::unordered_map byDevicePoint_; +}; + +} // namespace softbus::registry diff --git a/src/softbus_registry/include/softbus_registry/PointTableEntry.h b/src/softbus_registry/include/softbus_registry/PointTableEntry.h new file mode 100644 index 0000000..5bc60de --- /dev/null +++ b/src/softbus_registry/include/softbus_registry/PointTableEntry.h @@ -0,0 +1,41 @@ +#pragma once + +#include + +namespace softbus::registry { + +struct PointTableEntry { + std::string deviceId; + std::string pointId; + std::string name; + std::string displayName; + std::string objectRef; + std::string signalType; + std::string physicalChannel; + std::string unit; + double minValue{0.0}; + double maxValue{100.0}; + std::string access{"read"}; + bool writable{false}; + int registerAddress{0}; + double scale{1.0}; + std::string opcuaNodeId; + std::string cmdTopic; + std::string ackTopic; + std::string dataTopic; +}; + +struct TopicRules { + std::string data{"data/{deviceId}/{pointId}"}; + std::string command{"control/cmd/{deviceId}/{pointId}/set"}; + std::string ack{"control/ack/{deviceId}/{pointId}"}; + std::string event{"event/{deviceId}/{eventType}"}; + std::string heartbeat{"system/heartbeat/{deviceId}"}; + + std::string formatData(const std::string& deviceId, const std::string& pointId) const; + std::string formatCommand(const std::string& deviceId, const std::string& pointId) const; + std::string formatAck(const std::string& deviceId, const std::string& pointId) const; + std::string formatHeartbeat(const std::string& deviceId) const; +}; + +} // namespace softbus::registry diff --git a/src/softbus_registry/src/PointRegistry.cpp b/src/softbus_registry/src/PointRegistry.cpp new file mode 100644 index 0000000..63cdaac --- /dev/null +++ b/src/softbus_registry/src/PointRegistry.cpp @@ -0,0 +1,199 @@ +#include + +#include + +#include +#include + +namespace softbus::registry { + +bool PointRegistry::loadPointTable(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + + tree_.site = j.value("site", ""); + tree_.system = j.value("system", ""); + tree_.asset = j.value("asset", ""); + tree_.devices.clear(); + entries_.clear(); + byObjectRef_.clear(); + byTopic_.clear(); + byOpcuaNodeId_.clear(); + byDevicePoint_.clear(); + + if (!j.contains("devices")) { + return tree_.isValid(); + } + + for (const auto& dev : j["devices"]) { + softbus::core::DeviceDef device; + device.stableDeviceKey = dev.value("stableDeviceKey", dev.value("deviceId", "")); + device.protocol = dev.value("protocol", "modbus"); + const std::string deviceId = dev.value("deviceId", device.stableDeviceKey); + + if (!dev.contains("points")) { + tree_.devices.push_back(std::move(device)); + continue; + } + + for (const auto& pt : dev["points"]) { + softbus::core::PointDef pointDef; + pointDef.name = pt.value("name", pt.value("pointId", "")); + pointDef.registerAddress = pt.value("register", 0); + pointDef.domain = pt.value("domain", "Process"); + pointDef.signalType = pt.value("signalType", "AI"); + pointDef.physicalChannel = pt.value("physicalChannel", ""); + pointDef.unit = pt.value("unit", ""); + pointDef.scale = pt.value("scale", 1.0); + pointDef.access = pt.value("access", "read"); + pointDef.writable = pt.value("writable", false); + if (pt.contains("range") && pt["range"].is_array() && pt["range"].size() >= 2) { + pointDef.minValue = pt["range"][0].get(); + pointDef.maxValue = pt["range"][1].get(); + } + if (pointDef.access == "read_write" || pointDef.access == "write") { + pointDef.writable = true; + } + device.points.push_back(pointDef); + + PointTableEntry entry; + entry.deviceId = deviceId; + entry.pointId = pt.value("pointId", pointDef.name); + entry.name = pointDef.name; + entry.displayName = pt.value("displayName", entry.name); + entry.signalType = pointDef.signalType; + entry.physicalChannel = pointDef.physicalChannel; + entry.unit = pointDef.unit; + entry.minValue = pointDef.minValue; + entry.maxValue = pointDef.maxValue; + entry.access = pointDef.access; + entry.writable = pointDef.writable; + entry.registerAddress = pointDef.registerAddress; + entry.scale = pointDef.scale; + + const auto objectRef = softbus::core::DeviceIdentity::buildObjectRef( + tree_, device.stableDeviceKey, entry.name); + entry.objectRef = objectRef.toPath(); + + entry.opcuaNodeId = pt.value( + "opcuaNodeId", "ns=1;s=" + entry.objectRef); + entry.dataTopic = pt.value( + "dataTopic", topicRules_.formatData(entry.deviceId, entry.pointId)); + entry.cmdTopic = pt.value( + "cmdTopic", topicRules_.formatCommand(entry.deviceId, entry.pointId)); + entry.ackTopic = pt.value( + "ackTopic", topicRules_.formatAck(entry.deviceId, entry.pointId)); + + entries_.push_back(entry); + indexEntry(entries_.back()); + } + tree_.devices.push_back(std::move(device)); + } + return tree_.isValid() && !entries_.empty(); +} + +bool PointRegistry::loadTopicRules(const std::string& path) +{ + std::ifstream ifs(path); + if (!ifs) { + return false; + } + nlohmann::json j; + ifs >> j; + topicRules_.data = j.value("data", topicRules_.data); + topicRules_.command = j.value("command", topicRules_.command); + topicRules_.ack = j.value("ack", topicRules_.ack); + topicRules_.event = j.value("event", topicRules_.event); + topicRules_.heartbeat = j.value("heartbeat", topicRules_.heartbeat); + return true; +} + +void PointRegistry::indexEntry(const PointTableEntry& entry) +{ + const std::size_t idx = entries_.size() - 1; + byObjectRef_[entry.objectRef] = idx; + byOpcuaNodeId_[entry.opcuaNodeId] = idx; + byDevicePoint_[entry.deviceId + "/" + entry.pointId] = idx; + if (!entry.dataTopic.empty()) { + byTopic_[entry.dataTopic] = idx; + } + if (!entry.cmdTopic.empty()) { + byTopic_[entry.cmdTopic] = idx; + } + if (!entry.ackTopic.empty()) { + byTopic_[entry.ackTopic] = idx; + } +} + +std::optional PointRegistry::findByObjectRef(const std::string& objectRef) const +{ + const auto it = byObjectRef_.find(objectRef); + if (it == byObjectRef_.end()) { + return std::nullopt; + } + return entries_[it->second]; +} + +std::optional PointRegistry::findByTopic(const std::string& topic) const +{ + const auto it = byTopic_.find(topic); + if (it == byTopic_.end()) { + return std::nullopt; + } + return entries_[it->second]; +} + +std::optional PointRegistry::findByOpcuaNodeId(const std::string& nodeId) const +{ + const auto it = byOpcuaNodeId_.find(nodeId); + if (it == byOpcuaNodeId_.end()) { + return std::nullopt; + } + return entries_[it->second]; +} + +std::optional PointRegistry::findByDevicePoint(const std::string& deviceId, + const std::string& pointId) const +{ + const auto it = byDevicePoint_.find(deviceId + "/" + pointId); + if (it == byDevicePoint_.end()) { + return std::nullopt; + } + return entries_[it->second]; +} + +std::optional PointRegistry::findPointByRegister(int registerAddress) const +{ + for (const auto& entry : entries_) { + if (entry.registerAddress == registerAddress) { + return softbus::core::ObjectRef::fromPath(entry.objectRef); + } + } + return std::nullopt; +} + +bool PointRegistry::validateWriteValue(const PointTableEntry& entry, double value, + std::string* error) const +{ + if (!entry.writable) { + if (error) { + *error = "point is not writable"; + } + return false; + } + if (value < entry.minValue || value > entry.maxValue) { + if (error) { + *error = "value out of range [" + std::to_string(entry.minValue) + "," + + std::to_string(entry.maxValue) + "]"; + } + return false; + } + return true; +} + +} // namespace softbus::registry diff --git a/src/softbus_registry/src/PointTableEntry.cpp b/src/softbus_registry/src/PointTableEntry.cpp new file mode 100644 index 0000000..bb2c306 --- /dev/null +++ b/src/softbus_registry/src/PointTableEntry.cpp @@ -0,0 +1,45 @@ +#include + +namespace softbus::registry { + +namespace { + +std::string replaceToken(std::string pattern, const std::string& token, const std::string& value) +{ + const auto pos = pattern.find(token); + if (pos == std::string::npos) { + return pattern; + } + pattern.replace(pos, token.size(), value); + return pattern; +} + +} // namespace + +std::string TopicRules::formatData(const std::string& deviceId, const std::string& pointId) const +{ + std::string topic = data; + topic = replaceToken(topic, "{deviceId}", deviceId); + return replaceToken(topic, "{pointId}", pointId); +} + +std::string TopicRules::formatCommand(const std::string& deviceId, const std::string& pointId) const +{ + std::string topic = command; + topic = replaceToken(topic, "{deviceId}", deviceId); + return replaceToken(topic, "{pointId}", pointId); +} + +std::string TopicRules::formatAck(const std::string& deviceId, const std::string& pointId) const +{ + std::string topic = ack; + topic = replaceToken(topic, "{deviceId}", deviceId); + return replaceToken(topic, "{pointId}", pointId); +} + +std::string TopicRules::formatHeartbeat(const std::string& deviceId) const +{ + return replaceToken(heartbeat, "{deviceId}", deviceId); +} + +} // namespace softbus::registry diff --git a/src/softbus_storage/CMakeLists.txt b/src/softbus_storage/CMakeLists.txt new file mode 100644 index 0000000..863cc58 --- /dev/null +++ b/src/softbus_storage/CMakeLists.txt @@ -0,0 +1,8 @@ +softbus_package(softbus_storage) + +add_library(softbus_storage STATIC + src/StorageService.cpp +) + +softbus_export_include(softbus_storage "${CMAKE_CURRENT_SOURCE_DIR}/include") +target_link_libraries(softbus_storage PUBLIC softbus_core softbus_bus) diff --git a/src/softbus_storage/include/softbus_storage/StorageService.h b/src/softbus_storage/include/softbus_storage/StorageService.h new file mode 100644 index 0000000..02f8aa9 --- /dev/null +++ b/src/softbus_storage/include/softbus_storage/StorageService.h @@ -0,0 +1,33 @@ +#pragma once + +#include + +#include +#include +#include + +namespace softbus::storage { + +class StorageService { +public: + explicit StorageService(std::string dataDir = "data/logs"); + + void attachBus(softbus::bus::Bus* bus); + bool open(); + +private: + void onBusMessage(const softbus::bus::BusMessage& message); + void appendLine(const std::string& fileName, const std::string& line); + + std::string dataDir_; + softbus::bus::Bus* bus_{nullptr}; + softbus::bus::Bus::SubscriptionId dataSub_{0}; + softbus::bus::Bus::SubscriptionId ackSub_{0}; + softbus::bus::Bus::SubscriptionId eventSub_{0}; + std::mutex mutex_; + std::ofstream timeseries_; + std::ofstream commandLog_; + std::ofstream eventLog_; +}; + +} // namespace softbus::storage diff --git a/src/softbus_storage/src/StorageService.cpp b/src/softbus_storage/src/StorageService.cpp new file mode 100644 index 0000000..095bb03 --- /dev/null +++ b/src/softbus_storage/src/StorageService.cpp @@ -0,0 +1,70 @@ +#include + +#include + +namespace softbus::storage { + +namespace fs = std::filesystem; + +StorageService::StorageService(std::string dataDir) + : dataDir_(std::move(dataDir)) +{ +} + +bool StorageService::open() +{ + fs::create_directories(dataDir_); + timeseries_.open(dataDir_ + "/timeseries.jsonl", std::ios::app); + commandLog_.open(dataDir_ + "/command_log.jsonl", std::ios::app); + eventLog_.open(dataDir_ + "/event_log.jsonl", std::ios::app); + return timeseries_.is_open() && commandLog_.is_open() && eventLog_.is_open(); +} + +void StorageService::attachBus(softbus::bus::Bus* bus) +{ + bus_ = bus; + if (!bus_) { + return; + } + open(); + dataSub_ = bus_->subscribe("data/#", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); + ackSub_ = bus_->subscribe("control/ack/#", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); + eventSub_ = bus_->subscribe("event/#", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); +} + +void StorageService::onBusMessage(const softbus::bus::BusMessage& message) +{ + switch (message.type) { + case softbus::bus::BusPayloadType::DataPoint: + appendLine("timeseries", message.dataPoint.toJson().dump()); + break; + case softbus::bus::BusPayloadType::Ack: + appendLine("command", message.ack.toJson().dump()); + break; + case softbus::bus::BusPayloadType::Event: + appendLine("event", message.event.toJson().dump()); + break; + default: + break; + } +} + +void StorageService::appendLine(const std::string& fileName, const std::string& line) +{ + std::lock_guard lock(mutex_); + if (fileName == "timeseries" && timeseries_.is_open()) { + timeseries_ << line << '\n'; + } else if (fileName == "command" && commandLog_.is_open()) { + commandLog_ << line << '\n'; + } else if (fileName == "event" && eventLog_.is_open()) { + eventLog_ << line << '\n'; + } +} + +} // namespace softbus::storage diff --git a/src/softbus_transport/CMakeLists.txt b/src/softbus_transport/CMakeLists.txt index 04bdbb2..67b9a89 100644 --- a/src/softbus_transport/CMakeLists.txt +++ b/src/softbus_transport/CMakeLists.txt @@ -4,9 +4,10 @@ add_library(softbus_transport STATIC src/UplinkWire.cpp src/TcpUplinkServer.cpp src/TcpUplinkClient.cpp + src/TransportBusBridge.cpp ) softbus_export_include(softbus_transport "${CMAKE_CURRENT_SOURCE_DIR}/include") -target_link_libraries(softbus_transport PUBLIC softbus_core softbus_api asio) +target_link_libraries(softbus_transport PUBLIC softbus_core softbus_api softbus_bus asio) softbus_apply_platform_libs(softbus_transport) diff --git a/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h b/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h index 117337f..de0d783 100644 --- a/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h +++ b/src/softbus_transport/include/softbus_transport/TcpUplinkClient.h @@ -22,7 +22,10 @@ public: void setCommandHandler(CommandHandler handler); bool connect(); void disconnect(); + bool sendCommand(const softbus::api::ExecuteCommand& command); bool sendRawPacket(const softbus::core::RawPacket& packet); + bool sendAck(const softbus::core::SoftbusAck& ack); + bool sendHeartbeat(const std::string& deviceId); void poll(); bool isConnected() const { return connected_; } diff --git a/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h b/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h index d60653e..a9de07c 100644 --- a/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h +++ b/src/softbus_transport/include/softbus_transport/TcpUplinkServer.h @@ -14,6 +14,7 @@ namespace softbus::transport { using RawPacketHandler = std::function; +using WireMessageHandler = std::function; using CommandSender = std::function; class TcpUplinkServer { @@ -22,6 +23,7 @@ public: ~TcpUplinkServer(); void setRawPacketHandler(RawPacketHandler handler); + void setWireMessageHandler(WireMessageHandler handler); void start(); void stop(); bool sendCommand(const softbus::api::ExecuteCommand& command); @@ -34,6 +36,7 @@ private: asio::io_context& io_; asio::ip::tcp::acceptor acceptor_; RawPacketHandler rawHandler_; + WireMessageHandler wireHandler_; std::shared_ptr clientSocket_; std::string readBuffer_; std::mutex sendMutex_; diff --git a/src/softbus_transport/include/softbus_transport/TransportBusBridge.h b/src/softbus_transport/include/softbus_transport/TransportBusBridge.h new file mode 100644 index 0000000..c573ec5 --- /dev/null +++ b/src/softbus_transport/include/softbus_transport/TransportBusBridge.h @@ -0,0 +1,21 @@ +#pragma once + +#include +#include + +namespace softbus::transport { + +class TransportBusBridge { +public: + void attach(softbus::bus::Bus* bus, TcpUplinkServer* server); + +private: + void onBusMessage(const softbus::bus::BusMessage& message); + void onWireMessage(const WireMessage& message); + + softbus::bus::Bus* bus_{nullptr}; + TcpUplinkServer* server_{nullptr}; + softbus::bus::Bus::SubscriptionId cmdSub_{0}; +}; + +} // namespace softbus::transport diff --git a/src/softbus_transport/include/softbus_transport/UplinkWire.h b/src/softbus_transport/include/softbus_transport/UplinkWire.h index fb06b0d..c52cea4 100644 --- a/src/softbus_transport/include/softbus_transport/UplinkWire.h +++ b/src/softbus_transport/include/softbus_transport/UplinkWire.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include @@ -14,6 +15,8 @@ namespace softbus::transport { enum class WireMessageType { RawPacket, ExecuteCommand, + Ack, + Heartbeat, Unknown }; @@ -21,10 +24,14 @@ struct WireMessage { WireMessageType type{WireMessageType::Unknown}; softbus::core::RawPacket rawPacket; softbus::api::ExecuteCommand command; + softbus::core::SoftbusAck ack; + std::string heartbeatDeviceId; }; std::string encodeWireMessage(const softbus::core::RawPacket& packet); std::string encodeWireMessage(const softbus::api::ExecuteCommand& command); +std::string encodeWireMessage(const softbus::core::SoftbusAck& ack); +std::string encodeWireMessage(const std::string& deviceId); std::optional decodeWireMessage(const std::string& line); std::string base64Encode(const uint8_t* data, std::size_t size); diff --git a/src/softbus_transport/src/TcpUplinkClient.cpp b/src/softbus_transport/src/TcpUplinkClient.cpp index a546fa0..4e53d74 100644 --- a/src/softbus_transport/src/TcpUplinkClient.cpp +++ b/src/softbus_transport/src/TcpUplinkClient.cpp @@ -62,6 +62,42 @@ void TcpUplinkClient::disconnect() asio::make_work_guard(io_)); } +bool TcpUplinkClient::sendCommand(const softbus::api::ExecuteCommand& command) +{ + if (!socket_ || !socket_->is_open()) { + return false; + } + const auto line = encodeWireMessage(command); + std::lock_guard lock(sendMutex_); + std::error_code ec; + asio::write(*socket_, asio::buffer(line), ec); + return !ec; +} + +bool TcpUplinkClient::sendAck(const softbus::core::SoftbusAck& ack) +{ + if (!socket_ || !socket_->is_open()) { + return false; + } + const auto line = encodeWireMessage(ack); + std::lock_guard lock(sendMutex_); + std::error_code ec; + asio::write(*socket_, asio::buffer(line), ec); + return !ec; +} + +bool TcpUplinkClient::sendHeartbeat(const std::string& deviceId) +{ + if (!socket_ || !socket_->is_open()) { + return false; + } + const auto line = encodeWireMessage(deviceId); + std::lock_guard lock(sendMutex_); + std::error_code ec; + asio::write(*socket_, asio::buffer(line), ec); + return !ec; +} + bool TcpUplinkClient::sendRawPacket(const softbus::core::RawPacket& packet) { if (!socket_ || !socket_->is_open()) { diff --git a/src/softbus_transport/src/TcpUplinkServer.cpp b/src/softbus_transport/src/TcpUplinkServer.cpp index 83532ea..d46978f 100644 --- a/src/softbus_transport/src/TcpUplinkServer.cpp +++ b/src/softbus_transport/src/TcpUplinkServer.cpp @@ -22,6 +22,11 @@ void TcpUplinkServer::setRawPacketHandler(RawPacketHandler handler) rawHandler_ = std::move(handler); } +void TcpUplinkServer::setWireMessageHandler(WireMessageHandler handler) +{ + wireHandler_ = std::move(handler); +} + void TcpUplinkServer::start() { running_ = true; @@ -107,6 +112,9 @@ void TcpUplinkServer::handleLine(const std::string& line) if (!msg) { return; } + if (wireHandler_) { + wireHandler_(*msg); + } if (msg->type == WireMessageType::RawPacket && rawHandler_) { rawHandler_(msg->rawPacket); } diff --git a/src/softbus_transport/src/TransportBusBridge.cpp b/src/softbus_transport/src/TransportBusBridge.cpp new file mode 100644 index 0000000..feb5a4d --- /dev/null +++ b/src/softbus_transport/src/TransportBusBridge.cpp @@ -0,0 +1,70 @@ +#include + +#include + +namespace softbus::transport { + +void TransportBusBridge::attach(softbus::bus::Bus* bus, TcpUplinkServer* server) +{ + bus_ = bus; + server_ = server; + if (!bus_ || !server_) { + return; + } + + server_->setWireMessageHandler([this](const WireMessage& msg) { + onWireMessage(msg); + }); + + cmdSub_ = bus_->subscribe("control/cmd/#", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); + bus_->subscribe("control/cmd/injected", [this](const softbus::bus::BusMessage& msg) { + onBusMessage(msg); + }); +} + +void TransportBusBridge::onBusMessage(const softbus::bus::BusMessage& message) +{ + if (!server_ || message.type != softbus::bus::BusPayloadType::ExecuteCommand) { + return; + } + if (!server_->sendCommand(message.command)) { + SB_LOG_WARN("TransportBusBridge", "failed to send command to edge topic=", message.topic); + } +} + +void TransportBusBridge::onWireMessage(const WireMessage& message) +{ + if (!bus_) { + return; + } + + softbus::bus::BusMessage busMsg; + switch (message.type) { + case WireMessageType::RawPacket: + busMsg.type = softbus::bus::BusPayloadType::RawPacket; + busMsg.rawPacket = message.rawPacket; + bus_->publish("edge/raw/" + message.rawPacket.sourceId, busMsg); + break; + case WireMessageType::Ack: + busMsg.type = softbus::bus::BusPayloadType::Ack; + busMsg.ack = message.ack; + bus_->publish("control/ack/" + message.ack.deviceId + "/" + message.ack.pointId, busMsg); + break; + case WireMessageType::Heartbeat: + busMsg.type = softbus::bus::BusPayloadType::Heartbeat; + busMsg.deviceId = message.heartbeatDeviceId; + bus_->publish("system/heartbeat/" + message.heartbeatDeviceId, busMsg); + break; + case WireMessageType::ExecuteCommand: + busMsg.type = softbus::bus::BusPayloadType::ExecuteCommand; + busMsg.command = message.command; + bus_->publish("control/cmd/injected", busMsg); + break; + default: + break; + } +} + +} // namespace softbus::transport diff --git a/src/softbus_transport/src/UplinkWire.cpp b/src/softbus_transport/src/UplinkWire.cpp index 20482e1..a712a75 100644 --- a/src/softbus_transport/src/UplinkWire.cpp +++ b/src/softbus_transport/src/UplinkWire.cpp @@ -1,5 +1,6 @@ #include +#include #include namespace softbus::transport { @@ -78,6 +79,23 @@ std::string encodeWireMessage(const softbus::api::ExecuteCommand& command) return command.toJson().dump() + "\n"; } +std::string encodeWireMessage(const softbus::core::SoftbusAck& ack) +{ + return ack.toJson().dump() + "\n"; +} + +std::string encodeWireMessage(const std::string& deviceId) +{ + nlohmann::json j{ + {"type", "heartbeat"}, + {"deviceId", deviceId}, + {"timestampNs", static_cast( + std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count())} + }; + return j.dump() + "\n"; +} + std::optional decodeWireMessage(const std::string& line) { if (line.empty()) { @@ -104,6 +122,16 @@ std::optional decodeWireMessage(const std::string& line) msg.command = softbus::api::ExecuteCommand::fromJson(j); return msg; } + if (type == "ack") { + msg.type = WireMessageType::Ack; + msg.ack = softbus::core::SoftbusAck::fromJson(j); + return msg; + } + if (type == "heartbeat") { + msg.type = WireMessageType::Heartbeat; + msg.heartbeatDeviceId = j.value("deviceId", ""); + return msg; + } return std::nullopt; } diff --git a/tests/integration/valve_control_loop.sh b/tests/integration/valve_control_loop.sh new file mode 100755 index 0000000..ff8d462 --- /dev/null +++ b/tests/integration/valve_control_loop.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env bash +set -euo pipefail +ROOT="$(cd "$(dirname "$0")/../.." && pwd)" +cd "$ROOT" + +fuser -k 9000/tcp 4840/tcp 2>/dev/null || true +sleep 1 + +./build/bin/softbus_daemon \ + --config config/daemon_profile.json \ + --point-table config/point_table.json \ + --self-test > /tmp/valve_loop_daemon.log 2>&1 & +DAEMON_PID=$! +sleep 1 +./build/bin/softbus_edge --config config/edge.json > /tmp/valve_loop_edge.log 2>&1 & +EDGE_PID=$! +sleep 8 + +kill "$DAEMON_PID" "$EDGE_PID" 2>/dev/null || true +fuser -k 9000/tcp 4840/tcp 2>/dev/null || true + +grep -q "state=Succeeded" /tmp/valve_loop_daemon.log +grep -q "executedValue=60" /tmp/valve_loop_daemon.log +grep -q "ValveOpening value=60" /tmp/valve_loop_edge.log + +echo "valve_control_loop: PASS" diff --git a/tools/CMakeLists.txt b/tools/CMakeLists.txt new file mode 100644 index 0000000..8203880 --- /dev/null +++ b/tools/CMakeLists.txt @@ -0,0 +1,11 @@ +add_executable(cmd_sender cmd_sender.cpp) +target_link_libraries(cmd_sender PRIVATE + softbus_core softbus_api softbus_transport softbus_registry asio +) +softbus_apply_platform_libs(cmd_sender) + +add_executable(topic_watcher topic_watcher.cpp) +target_link_libraries(topic_watcher PRIVATE + softbus_bus softbus_core +) +softbus_apply_platform_libs(topic_watcher) diff --git a/tools/cmd_sender.cpp b/tools/cmd_sender.cpp new file mode 100644 index 0000000..269aa1d --- /dev/null +++ b/tools/cmd_sender.cpp @@ -0,0 +1,45 @@ +#include +#include +#include + +#include +#include + +int main(int argc, char* argv[]) +{ + std::string host = "127.0.0.1"; + uint16_t port = 9000; + std::string objectRef = "DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening"; + double value = 60.0; + + for (int i = 1; i < argc; ++i) { + const std::string arg = argv[i]; + if (arg == "--daemon" && i + 1 < argc) { + const std::string endpoint = argv[++i]; + const auto pos = endpoint.find(':'); + if (pos != std::string::npos) { + host = endpoint.substr(0, pos); + port = static_cast(std::stoi(endpoint.substr(pos + 1))); + } + } else if (arg == "--object-ref" && i + 1 < argc) { + objectRef = argv[++i]; + } else if (arg == "--value" && i + 1 < argc) { + value = std::stod(argv[++i]); + } + } + + softbus::api::ExecuteCommand cmd; + cmd.requestId = "req-cli-" + softbus::core::DeviceIdentity::generateTraceId(); + cmd.traceId = softbus::core::DeviceIdentity::generateTraceId(); + cmd.objectRef = objectRef; + cmd.command = "write"; + cmd.params = {{"value", value}}; + + std::cout << cmd.toJson().dump() << std::endl; + std::cout << "Use OPC UA ExecuteCommand or: softbus_daemon --self-test" << std::endl; + std::cout << "Note: uplink port " << host << ":" << port + << " is reserved for edge agent connection." << std::endl; + (void)host; + (void)port; + return 0; +} diff --git a/tools/topic_watcher.cpp b/tools/topic_watcher.cpp new file mode 100644 index 0000000..11eb129 --- /dev/null +++ b/tools/topic_watcher.cpp @@ -0,0 +1,15 @@ +#include + +#include +#include + +int main() +{ + auto& bus = softbus::bus::Bus::instance(); + bus.subscribe("#", [](const softbus::bus::BusMessage& msg) { + std::cout << "[topic] " << msg.topic << std::endl; + }); + std::cout << "topic_watcher: attach to daemon process is standalone demo only." << std::endl; + std::cout << "Use daemon logs or integrate watcher via shared bus in-process." << std::endl; + return 0; +}