phase 2 serial

This commit is contained in:
flower_linux
2026-06-12 18:34:49 +08:00
parent 0cf0713dc5
commit 628ccf1409
78 changed files with 2571 additions and 345 deletions

View File

@@ -1,5 +1,5 @@
cmake_minimum_required(VERSION 3.16)
project(softbus_workspace LANGUAGES CXX)
project(softbus_workspace LANGUAGES CXX C)
set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)

View File

@@ -73,6 +73,45 @@ else()
set(SOFTBUS_HAS_OPCUA OFF)
endif()
# libmodbus (edge RTU/TCP): system pkg-config first, else third_party vendor
set(SOFTBUS_HAS_LIBMODBUS OFF)
set(SOFTBUS_LIBMODBUS_VENDOR OFF)
option(SOFTBUS_FORCE_VENDOR_LIBMODBUS
"Always build libmodbus from third_party/libmodbus" OFF)
if(NOT SOFTBUS_FORCE_VENDOR_LIBMODBUS)
find_package(PkgConfig QUIET)
if(PkgConfig_FOUND)
pkg_check_modules(LIBMODBUS QUIET libmodbus)
endif()
endif()
if(LIBMODBUS_FOUND AND NOT SOFTBUS_FORCE_VENDOR_LIBMODBUS)
set(SOFTBUS_HAS_LIBMODBUS ON)
add_library(libmodbus_iface INTERFACE)
target_include_directories(libmodbus_iface INTERFACE ${LIBMODBUS_INCLUDE_DIRS})
target_link_libraries(libmodbus_iface INTERFACE ${LIBMODBUS_LIBRARIES})
message(STATUS "libmodbus: using system package (pkg-config)")
else()
add_subdirectory("${CMAKE_CURRENT_LIST_DIR}/libmodbus")
if(NOT SOFTBUS_HAS_LIBMODBUS)
message(STATUS "libmodbus not available; softbus_edge uses simulated Modbus I/O")
endif()
endif()
function(softbus_apply_libmodbus target)
if(SOFTBUS_HAS_LIBMODBUS)
target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_LIBMODBUS=1)
target_link_libraries(${target} PRIVATE libmodbus_iface)
if(SOFTBUS_LIBMODBUS_VENDOR)
target_compile_definitions(${target} PRIVATE SOFTBUS_LIBMODBUS_VENDOR=1)
endif()
else()
target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_LIBMODBUS=0)
endif()
endfunction()
function(softbus_apply_open62541 target)
if(SOFTBUS_HAS_OPCUA)
target_compile_definitions(${target} PRIVATE SOFTBUS_HAS_OPCUA=1)

View File

@@ -0,0 +1,51 @@
# Vendored libmodbus v3.1.10 static library for softbus_edge
set(_src_dir "${CMAKE_SOURCE_DIR}/third_party/libmodbus")
if(NOT EXISTS "${_src_dir}/src/modbus.c")
message(WARNING "Vendor libmodbus not found at ${_src_dir}")
return()
endif()
set(_gen_dir "${CMAKE_BINARY_DIR}/libmodbus_vendor")
file(MAKE_DIRECTORY "${_gen_dir}")
configure_file(
"${CMAKE_SOURCE_DIR}/cmake/libmodbus/config.h.linux"
"${_gen_dir}/config.h"
COPYONLY
)
configure_file(
"${CMAKE_SOURCE_DIR}/cmake/libmodbus/modbus-version.h"
"${_gen_dir}/modbus-version.h"
COPYONLY
)
add_library(modbus_vendor STATIC
"${_src_dir}/src/modbus.c"
"${_src_dir}/src/modbus-data.c"
"${_src_dir}/src/modbus-rtu.c"
"${_src_dir}/src/modbus-tcp.c"
)
set_target_properties(modbus_vendor PROPERTIES
C_STANDARD 99
POSITION_INDEPENDENT_CODE ON
)
target_include_directories(modbus_vendor
PUBLIC
"${_src_dir}/src"
"${_gen_dir}"
)
if(UNIX)
target_link_libraries(modbus_vendor PUBLIC pthread)
endif()
add_library(libmodbus_iface INTERFACE)
target_link_libraries(libmodbus_iface INTERFACE modbus_vendor)
target_include_directories(libmodbus_iface INTERFACE
"${_src_dir}/src"
"${_gen_dir}"
)
set(SOFTBUS_HAS_LIBMODBUS ON PARENT_SCOPE)
set(SOFTBUS_LIBMODBUS_VENDOR ON PARENT_SCOPE)
message(STATUS "libmodbus: using vendored build from third_party/libmodbus")

View File

@@ -0,0 +1,46 @@
/* Linux config.h for vendored libmodbus v3.1.10 */
#define HAVE_ARPA_INET_H 1
#define HAVE_DECL_TIOCSRS485 1
#define HAVE_DECL_TIOCM_RTS 1
#define HAVE_ERRNO_H 1
#define HAVE_FCNTL_H 1
#define HAVE_FORK 1
#define HAVE_GETADDRINFO 1
#define HAVE_GETTIMEOFDAY 1
#define HAVE_INTTYPES_H 1
#define HAVE_LIMITS_H 1
#define HAVE_LINUX_SERIAL_H 1
#define HAVE_MEMORY_H 1
#define HAVE_MEMSET 1
#define HAVE_NETDB_H 1
#define HAVE_NETINET_IN_H 1
#define HAVE_NETINET_TCP_H 1
#define HAVE_SELECT 1
#define HAVE_SOCKET 1
#define HAVE_STDINT_H 1
#define HAVE_STDLIB_H 1
#define HAVE_STRERROR 1
#define HAVE_STRINGS_H 1
#define HAVE_STRING_H 1
#define HAVE_SYS_IOCTL_H 1
#define HAVE_SYS_SOCKET_H 1
#define HAVE_SYS_STAT_H 1
#define HAVE_SYS_TIME_H 1
#define HAVE_SYS_TYPES_H 1
#define HAVE_TERMIOS_H 1
#define HAVE_TIME_H 1
#define HAVE_UNISTD_H 1
#define HAVE_VFORK 1
#define HAVE_WORKING_FORK 1
#define HAVE_WORKING_VFORK 1
#define PACKAGE "libmodbus"
#define PACKAGE_BUGREPORT "https://github.com/stephane/libmodbus/issues"
#define PACKAGE_NAME "libmodbus"
#define PACKAGE_STRING "libmodbus 3.1.10"
#define PACKAGE_TARNAME "libmodbus"
#define PACKAGE_URL ""
#define PACKAGE_VERSION "3.1.10"
#define STDC_HEADERS 1
#define TIME_WITH_SYS_TIME 1
#define VERSION "3.1.10"

View File

@@ -0,0 +1,19 @@
#ifndef MODBUS_VERSION_H
#define MODBUS_VERSION_H
#define LIBMODBUS_VERSION_MAJOR (3)
#define LIBMODBUS_VERSION_MINOR (1)
#define LIBMODBUS_VERSION_MICRO (10)
#define LIBMODBUS_VERSION 3.1.10
#define LIBMODBUS_VERSION_STRING "3.1.10"
#define LIBMODBUS_VERSION_HEX \
((LIBMODBUS_VERSION_MAJOR << 16) | (LIBMODBUS_VERSION_MINOR << 8) | \
(LIBMODBUS_VERSION_MICRO << 0))
#define LIBMODBUS_VERSION_CHECK(major, minor, micro) \
(LIBMODBUS_VERSION_MAJOR > (major) || \
(LIBMODBUS_VERSION_MAJOR == (major) && LIBMODBUS_VERSION_MINOR > (minor)) || \
(LIBMODBUS_VERSION_MAJOR == (major) && LIBMODBUS_VERSION_MINOR == (minor) && \
LIBMODBUS_VERSION_MICRO >= (micro)))
#endif /* MODBUS_VERSION_H */

View File

@@ -0,0 +1,20 @@
{
"bindings": [
{
"bindingId": "bind_temp_01",
"edgeId": "edge01",
"physicalPointId": "edge01/phy_modbus_s1_r40001",
"logicalPointId": "EdgeDevice_01/MotorTemperature",
"scale": 0.1,
"enabled": false
},
{
"bindingId": "bind_valve_01",
"edgeId": "edge01",
"physicalPointId": "edge01/phy_modbus_s1_r40002",
"logicalPointId": "EdgeDevice_01/ValveOpening",
"scale": 1.0,
"enabled": true
}
]
}

View File

@@ -0,0 +1,37 @@
{
"site": "DemoSite",
"system": "DemoSystem",
"asset": "DemoAsset",
"devices": [
{
"logicalDeviceId": "EdgeDevice_01",
"displayName": "边缘设备01",
"points": [
{
"logicalPointId": "MotorTemperature",
"displayName": "电机温度",
"signalType": "AI",
"unit": "℃",
"range": [0, 200],
"access": "read",
"writable": false,
"opcuaNodeId": "ns=1;s=DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature",
"dataTopic": "data/EdgeDevice_01/MotorTemperature"
},
{
"logicalPointId": "ValveOpening",
"displayName": "阀门开度",
"signalType": "AO",
"unit": "%",
"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"
}
]
}
]
}

View File

@@ -1,9 +0,0 @@
{
"edge": {
"deviceId": "EdgeDevice_01",
"daemonHost": "127.0.0.1",
"daemonPort": 9000,
"heartbeatPeriodMs": 1000,
"pointTable": "config/point_table.json"
}
}

View File

@@ -0,0 +1,17 @@
{
"edgeId": "edge01",
"bindings": [
{
"bindingId": "bind_temp_01",
"physicalPointId": "edge01/phy_modbus_s1_r40001",
"logicalPointId": "EdgeDevice_01/MotorTemperature",
"enabled": false
},
{
"bindingId": "bind_valve_01",
"physicalPointId": "edge01/phy_modbus_s1_r40002",
"logicalPointId": "EdgeDevice_01/ValveOpening",
"enabled": true
}
]
}

13
config/edge/edge.json Normal file
View File

@@ -0,0 +1,13 @@
{
"edge": {
"edgeId": "edge01",
"deviceId": "EdgeDevice_01",
"daemonHost": "127.0.0.1",
"daemonPort": 9000,
"heartbeatPeriodMs": 1000,
"uplinkReconnectPeriodMs": 5000,
"physicalChannels": "physical_channels.json",
"physicalDevices": "physical_devices.json",
"bindingsLocal": "bindings_local.json"
}
}

View File

@@ -0,0 +1,15 @@
{
"channels": [
{
"channelId": "serial_0",
"protocol": "modbus",
"transport": "rtu",
"port": "/dev/ttyS3",
"baudRate": 9600,
"parity": "N",
"dataBits": 8,
"stopBits": 1,
"pollIntervalMs": 500
}
]
}

View File

@@ -0,0 +1,29 @@
{
"devices": [
{
"physicalDeviceId": "phy_modbus_slave_1",
"channelId": "serial_0",
"protocol": "modbus",
"slaveId": 1,
"points": [
{
"physicalPointId": "phy_modbus_s1_r40001",
"register": 40001,
"function": 3,
"scale": 0.1,
"signalType": "AI",
"writable": false
},
{
"physicalPointId": "phy_modbus_s1_r40002",
"register": 40002,
"function": 3,
"writeFunction": 6,
"scale": 1.0,
"signalType": "AO",
"writable": true
}
]
}
]
}

60
docs/edge_device_setup.md Normal file
View File

@@ -0,0 +1,60 @@
# Edge 设备配置指南
## 配置目录
| 目录 | 用途 |
|------|------|
| `config/edge/` | 边缘设备本地配置(物理层 + 可选本地绑定子集) |
| `config/daemon/` | 核心守护进程配置逻辑点、全局绑定、OPC 主题规则) |
## 双域配置
| 文件 | 目录 | 职责 |
|------|------|------|
| `edge.json` | `config/edge/` | edge 入口uplink 地址、子配置文件路径 |
| `physical_channels.json` | `config/edge/` | 串口/网络接口 |
| `physical_devices.json` | `config/edge/` | 从站、寄存器、物理点 ID |
| `bindings_local.json` | `config/edge/` | 可选,本 edge 绑定子集(校验/本地调试) |
| `logical_points.json` | `config/daemon/` | OPC/业务语义 |
| `bindings_global.json` | `config/daemon/` | 物理点 ↔ 逻辑点 + edgeId 路由 |
| `topic_rule.json` | `config/daemon/` | 总线主题模板 |
| `point_table.json` | `config/daemon/` | **遗留**全量点表,仅 daemon 兼容加载 |
edge **不加载** `logical_points.json``point_table.json` 或全量 `bindings_global`;下行命令由 daemon 解析后携带 `physicalPointId``edgeId`
## 已知设备流程
1. edge编辑 `config/edge/physical_channels.json``config/edge/physical_devices.json`
2. daemon编辑 `config/daemon/logical_points.json`
3. daemon编辑 `config/daemon/bindings_global.json`(含 `edgeId``scale`
4. 启动:`softbus_daemon --logical-points config/daemon/logical_points.json --bindings-global config/daemon/bindings_global.json`
5. 启动:`softbus_edge --config config/edge/edge.json`
6. 热加载:`kill -HUP <daemon-pid>``kill -HUP <edge-pid>`
## 未知设备(物理先行)
1. 仅配置 `physical_devices.json`,物理点 ID 如 `phy_modbus_s1_r40001`
2. edge 启动后 daemon 收到 `edge/raw/unbound/*` 主题数据
3. 后补 `logical_points.json``bindings_global.json`HUP daemon 即可打通 OPC
## libmodbus
构建时自动选择:
1. **系统包**(优先):`sudo apt install libmodbus-dev`
2. **vendor**(默认回退):`third_party/libmodbus` v3.1.10 静态编译进 `softbus_edge`
强制使用 vendor
```bash
cmake -B build -DSOFTBUS_FORCE_VENDOR_LIBMODBUS=ON
cmake --build build
```
串口权限:`sudo usermod -aG dialout $USER` 后重新登录。
## Mock 模式
```bash
./build/bin/softbus_edge --config config/edge/edge.json --mock
```

View File

@@ -6,27 +6,177 @@
## 1. 概述
`softbus_edge` 是边缘代理可执行文件负责设备数据采集ingress、原始包上行egress与下行命令接收。Phase 1 使用 `MockModbusIngress` 模拟 Modbus 数据,无真实硬件依赖。
`softbus_edge` 是边缘代理进程,运行在靠近现场设备的一侧,负责:
## 2. CONSTRAINTS
- **物理域采集ingress**:按 `physical_*.json` 轮询 Modbus RTU/TCP 等设备,产出 `RawPacket`
- **上行egress/uplink**:经 TCP 将 `RawPacket`、心跳、注册信息发送给 `softbus_daemon`
- **下行egress**:接收 daemon 下发的 `ExecuteCommand`写回物理设备Modbus 写寄存器等)
- 边缘代理不得直接操作 OPC UA 或插件解析,仅产生 `RawPacket` 并发送
- ingress 实现须遵循 `IIngressPort` 接口,便于 Phase 2 替换为真实硬件驱动。
- 下行命令 Phase 1 仅日志输出,不得假装已写回设备。
- 边缘代理通过 CLI 参数配置连接地址,不读取 `daemon_profile.json`
edge **不**加载逻辑点表、**不**操作 OPC UA、**不**运行 `protocol_modbus` 插件。协议解析与 OPC 绑定均在 daemon 侧完成
## 3. 组件架构
| 模式 | 说明 |
|------|------|
| **正常模式** | libmodbus + `ModbusIngress`,读真实硬件 |
| **Mock 模式** | `--mock`,无硬件,模拟 Modbus 帧上行 |
| **独立模式** | daemon 未启动时 edge 仍可运行,后台重连 uplink |
---
## 2. 与 daemon 的职责边界(双域)
```
┌─────────────────────────────────────────────────────────────────┐
│ edge物理域 daemon逻辑域
├─────────────────────────────────────────────────────────────────┤
│ config/edge/ config/daemon/ │
│ edge.json daemon_profile.json │
│ physical_channels.json logical_points.json │
│ physical_devices.json bindings_global.json │
│ bindings_local.json可选 topic_rule.json │
│ │
│ 读硬件 → RawPacket RawPacket → 插件解析 │
│ 写硬件 ← physicalPointId → bindings → OPC UA 节点 │
└─────────────────────────────────────────────────────────────────┘
```
| 配置 | edge 是否加载 | 用途 |
|------|---------------|------|
| `physical_channels.json` | 是 | 串口 / TCP 通道参数 |
| `physical_devices.json` | 是 | 从站、寄存器、物理点 ID |
| `bindings_local.json` | 可选 | 下行写命令的**本地备用**解析(正常联调几乎不用) |
| `logical_points.json` | **否** | daemon 专用 |
| `bindings_global.json` | **否** | daemon 专用(上行绑定 + 下行路由) |
edge 上行时**不做** binding 初筛:所有已配置物理点都会采集并上报;是否绑定 OPC 由 daemon 的 `bindings_global.json` 决定(未绑定点进入 `edge/unbound/raw/...`)。
---
## 3. 配置加载流程
### 3.1 入口文件
默认配置:`config/edge/edge.json`。子路径**相对 `edge.json` 所在目录**解析。
```json
{
"edge": {
"edgeId": "edge01",
"deviceId": "EdgeDevice_01",
"daemonHost": "127.0.0.1",
"daemonPort": 9000,
"heartbeatPeriodMs": 1000,
"uplinkReconnectPeriodMs": 5000,
"physicalChannels": "physical_channels.json",
"physicalDevices": "physical_devices.json",
"bindingsLocal": "bindings_local.json"
}
}
```
| 字段 | 说明 |
|------|------|
| `edgeId` | 边缘实例 IDuplink 注册与 daemon 路由用 |
| `deviceId` | 逻辑设备名ACK 等场景) |
| `daemonHost` / `daemonPort` | 核心 uplink 地址 |
| `heartbeatPeriodMs` | 心跳周期(仅 uplink 已连接时发送) |
| `uplinkReconnectPeriodMs` | daemon 不可达时的重连间隔 |
| `physicalChannels` / `physicalDevices` | 物理层配置 |
| `bindingsLocal` | 可选;空则跳过 |
| `bindingsPatch` | 可选;热加载绑定补丁 |
### 3.2 启动时加载顺序(`EdgeRuntime::init`
```
1. EdgeConfig.loadFromFile(--config)
2. PhysicalRegistry.loadChannels(physical_channels.json)
3. PhysicalRegistry.loadDevices(physical_devices.json, edgeId)
→ 物理点 ID 自动加前缀edge01/phy_modbus_s1_r40001
4. [可选] BindingRegistry.loadLocal(bindings_local.json)
5. UplinkClient + CommandExecutor 初始化
6. DeviceFactory.createIngress → ModbusIngress 或 MockModbusIngress
7. ModbusChannelManager.openChannel逐通道 libmodbus 连接)
8. uplink.connect();成功则 sendEdgeRegister(edgeId)
→ 失败仅 WARNedge 独立运行并后台重连
```
### 3.3 物理通道(`physical_channels.json`
```json
{
"channels": [
{
"channelId": "serial_0",
"protocol": "modbus",
"transport": "rtu",
"port": "/dev/ttyS3",
"baudRate": 9600,
"parity": "N",
"dataBits": 8,
"stopBits": 1,
"pollIntervalMs": 500
}
]
}
```
| 字段 | 说明 |
|------|------|
| `transport` | `rtu`(串口)或 `tcp`Modbus TCP`host` + `tcpPort` |
| `pollIntervalMs` | 该通道下所有点的轮询周期(当前整通道统一) |
### 3.4 物理设备与点(`physical_devices.json`
```json
{
"devices": [
{
"physicalDeviceId": "phy_modbus_slave_1",
"channelId": "serial_0",
"protocol": "modbus",
"slaveId": 1,
"points": [
{
"physicalPointId": "phy_modbus_s1_r40001",
"register": 40001,
"function": 3,
"scale": 0.1,
"signalType": "AI",
"writable": false
}
]
}
]
}
```
运行时 `physicalPointId` 会变为 `{edgeId}/{physicalPointId}`,作为 `RawPacket.sourceId` 上报。
### 3.5 热加载
| 信号 | 行为 |
|------|------|
| `SIGHUP` | `reloadPhysical()`:重载 channels + devices重建 Modbus 连接 |
| `SIGHUP` | `reloadBindingsPatch()`:若配置了 `bindingsPatch` 则增量应用 |
| 文件变更 | `ConfigWatcher` 监视 `physical_channels``physical_devices``bindingsPatch` |
---
## 4. 组件架构
```
main.cpp
── MockModbusIngress (IIngressPort) ← 入站采集
├── UplinkClient (egress) ← TCP 上行/下行
├── DeviceFactory ← 运行时设备 ID 生成
└── DeviceDiscovery ← 设备发现Phase 1 stub
── EdgeRuntime
├── PhysicalRegistry ← physical_*.json
├── BindingRegistry (local) ← bindings_local.json可选
├── ModbusChannelManager ← libmodbus RTU/TCP I/O
├── IIngressPort
│ ├── ModbusIngress ← 正常模式
│ └── MockModbusIngress ← --mock
├── ModbusEgress ← 下行写寄存器
├── CommandExecutor ← ExecuteCommand → Modbus 写 / mock
└── UplinkClient ← TcpUplinkClient 封装
```
## 4. 核心组件
### 4.1 IIngressPort — 入站抽象
| 方法 | 说明 |
@@ -35,102 +185,252 @@ main.cpp
| `close()` | 关闭采集通道 |
| `poll()` | 轮询,返回 `vector<RawPacket>` |
**Phase 1 实现:** `MockModbusIngress`
**ModbusIngress**:按 `pollIntervalMs` 遍历 `pollPoints()`,经 `ModbusChannelManager` 读寄存器,组装 Modbus 0x03 响应帧作为 `payload`
- 生成递增的 Modbus RTU 0x03 响应帧
- 寄存器值按 0.1 缩放(与 `ModbusParser` 一致)
- `sourceId` 使用 CLI 指定的 objectRef
**MockModbusIngress**:无硬件,生成递增值模拟帧。
### 4.2 UplinkClient — 上行客户端
### 4.2 ModbusChannelManager — 硬件 I/O
`TcpUplinkClient` 的薄封装:
- 构建时链接 **libmodbus**(系统包或 `third_party/libmodbus` vendor
- RTU`modbus_new_rtu` + `modbus_read_registers` / `modbus_write_register`
- TCP`modbus_new_tcp``transport: tcp` 时)
- 无 libmodbus 时退化为模拟读写(编译宏 `SOFTBUS_HAS_LIBMODBUS=0`
> edge **不使用** daemon 侧的 `protocol_modbus.so` 插件;插件仅用于 daemon 解析上行 RawPacket。
### 4.3 UplinkClient — 上行/下行
| 方法 | 说明 |
|------|------|
| `connect(host, port)` | 连接守护进程 |
| `sendRawPacket(RawPacket)` | 发送上行原始包 |
| `poll()` | 轮询接收下行命令 |
| `connect()` / `disconnect()` | TCP 连接 daemon |
| `sendEdgeRegister(edgeId)` | 注册边缘实例 |
| `sendHeartbeat(deviceId)` | 心跳 |
| `sendRawPacket(RawPacket)` | 上行原始包 |
| `sendAck(SoftbusAck)` | 写命令 ACK |
| `poll()` | 接收下行 `ExecuteCommand` |
| `setCommandHandler(handler)` | 注册命令回调 |
### 4.3 DeviceFactory
协议详见 [UPLINK_WIRE_PROTOCOL.md](UPLINK_WIRE_PROTOCOL.md)。
| 方法 | 说明 |
|------|------|
| `createRuntimeDevice()` | 生成 `runtime-xxxxxxxx` ID |
### 4.4 CommandExecutor — 下行写
### 4.4 DeviceDiscoveryPhase 1 stub
daemon 经 `bindings_global` 在命令中注入 `physicalPointId` + `edgeId`edge 直接写 Modbus
| 方法 | 说明 | Phase 1 行为 |
|------|------|-------------|
| `discover()` | 扫描可用设备 | 返回空列表 |
```
ExecuteCommand.params.physicalPointId → PhysicalRegistry → ModbusEgress.write
```
## 5. 主循环
仅当命令**缺少** `physicalPointId` 时,才用 `bindings_local.json``logicalPointId → physicalPointId` 备用解析。
---
## 5. 数据流
### 5.1 上行(采集 → OPC
```mermaid
sequenceDiagram
participant HW as 现场设备
participant Edge as softbus_edge
participant Daemon as softbus_daemon
participant OPC as OPC UA
HW->>Edge: Modbus RTU/TCP 读寄存器
Edge->>Edge: ModbusIngress.poll → RawPacket
Note over Edge: sourceId = edge01/phy_xxx<br/>protocolTag = modbus
Edge->>Daemon: TCP rawPacket + edgeRegister/heartbeat
Daemon->>Daemon: protocol_modbus 解析帧
Daemon->>Daemon: bindings_global → logicalPointId
Daemon->>Daemon: logical_points → objectRef + scale
Daemon->>OPC: UaModelBinder 写节点值
```
逐步说明:
1. **edge**`ModbusIngress` 读硬件 → `RawPacket{ sourceId=physicalPointId, payload=Modbus帧 }`
2. **edge**`UplinkClient.sendRawPacket` → TCP JSON 行
3. **daemon**`TransportBusBridge` 发布到总线 `edge/raw/{physicalPointId}`
4. **daemon**`PipelineEngine` + `protocol_modbus` 插件解析 → 寄存器值
5. **daemon**`bindings_global` 查绑定 → `logical_points` 得 OPC 路径 → scale → 写 OPC UA
未绑定物理点daemon 发布到 `edge/unbound/raw/{physicalPointId}`,不写 OPC。
### 5.2 下行OPC 写 → 设备)
```mermaid
sequenceDiagram
participant OPC as OPC UA 客户端
participant Daemon as softbus_daemon
participant Edge as softbus_edge
participant HW as 现场设备
OPC->>Daemon: ExecuteCommand (objectRef + value)
Daemon->>Daemon: bindings_global → physicalPointId + edgeId
Daemon->>Edge: TCP executeCommand
Edge->>Edge: CommandExecutor → ModbusEgress
Edge->>HW: libmodbus 写寄存器
Edge->>Daemon: TCP ack
```
### 5.3 主循环(`main.cpp`
```cpp
while (running) {
// 1. 采集
auto packets = ingress.poll();
for (auto& packet : packets) {
uplink.sendRawPacket(packet);
}
while (!shouldStop) {
ConfigWatcher.poll();
if (SIGHUP) { reloadPhysical(); reloadBindingsPatch(); }
// 2. 接收命令
uplink.poll();
runtime.tick(); // uplink.poll + ingress.poll + sendRawPacket
// 3. 平台信号检查
if (PlatformSignal::interrupted()) break;
if (uplinkConnected && heartbeatDue)
sendHeartbeat(edgeId);
sleep(50ms);
}
```
---
## 6. CLI 参数
| 参数 | 默认值 | 说明 |
|------|--------|------|
| `--daemon host:port` | `127.0.0.1:9000` | 守护进程地址 |
| `--object-ref path` | `DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature` | 模拟数据源对象路径 |
| `--config <path>` | `config/edge/edge.json` | 边缘入口配置 |
| `--mock` | 关闭 | 使用 MockModbusIngress跳过 physical 配置与 libmodbus |
| `--object-ref <path>` | `DemoSite/.../MotorTemperature` | Mock 模式下的模拟 sourceId |
## 7. 数据流
> 不再使用 `--daemon host:port`uplink 地址由 `edge.json` 的 `daemonHost` / `daemonPort` 指定。
### 7.1 上行
---
```
MockModbusIngress.poll()
→ RawPacket(sourceId=objectRef, protocolTag="modbus", payload=Modbus frame)
→ UplinkClient.sendRawPacket()
→ TcpUplinkClient → TCP JSON → Daemon
```
## 7. 依赖
### 7.2 下行
| 依赖 | 用途 |
|------|------|
| `softbus_core` | RawPacket、Logger、PlatformSignal、ConfigWatcher |
| `softbus_registry` | BindingRegistry仅 local 可选) |
| `softbus_transport` | TcpUplinkClient、UplinkWire |
| `softbus_api` | ExecuteCommand、SoftbusAck |
| **libmodbus** | Modbus RTU/TCP 硬件 I/Ovendor 或系统包) |
| asio | TcpUplinkClient 底层 |
```
Daemon CommandDispatcher
→ TCP JSON → TcpUplinkClient
→ UplinkClient.poll() → commandHandler
→ Phase 1: Logger 输出命令内容
```
---
## 8. 依赖
## 8. 构建与运行
- `softbus_core`RawPacket, DeviceIdentity, Logger, PlatformSignal
- `softbus_transport`TcpUplinkClient, UplinkWire
- `softbus_api`ExecuteCommand
- `asio`
### 8.1 libmodbus
## 9. 运行示例
构建时自动选择:系统 `libmodbus-dev` 优先,否则编译 `third_party/libmodbus` 静态库进 edge。
```bash
# 确保守护进程已启动
./build/bin/softbus_edge --daemon 127.0.0.1:9000
# 指定自定义对象路径
./build/bin/softbus_edge \
--daemon 127.0.0.1:9000 \
--object-ref DemoSite/DemoSystem/DemoAsset/SimModbusDevice/Temperature
cmake -B build
cmake --build build --target softbus_edge
```
## 10. Phase 2 规划
串口权限:`sudo usermod -aG dialout $USER`
- 实现真实 `SerialIngress` / `CanIngress` 替换 Mock
- 下行 `ExecuteCommand` 写回设备Modbus 写寄存器等)
- 设备发现与动态注册
- 支持 SBUP 二进制协议
### 8.2 运行示例
```bash
# 1. 先启动 daemon
./build/bin/softbus_daemon \
--config config/daemon/daemon_profile.json \
--logical-points config/daemon/logical_points.json \
--bindings-global config/daemon/bindings_global.json
# 2. 启动 edge真实 Modbus
./build/bin/softbus_edge --config config/edge/edge.json
# 3. 无硬件 / 无 daemon 调试
./build/bin/softbus_edge --config config/edge/edge.json --mock
# 4. 热加载物理配置
kill -HUP $(pidof softbus_edge)
```
更多配置说明见 [edge_device_setup.md](edge_device_setup.md)。
---
## 9. 未来扩展CANopen 与以太网设备
当前仅 **Modbus RTU/TCP** 在 edge 侧有完整 ingress/egress。CANopen 与更多以太网协议按同一双域模型扩展。
### 9.1 目标架构
```
physical_channels.json ──→ ChannelDriver按 protocol 选择)
physical_devices.json ──→ PhysicalRegistry
┌───────────┼───────────┐
▼ ▼ ▼
ModbusIngress CanIngress EthIngress规划
│ │ │
└───────────┴───────────┘
RawPacket → UplinkClient
```
daemon 侧已有对应解析插件(`protocol_modbus``protocol_canopen`edge 只需换 ingress/driver**上行仍是 RawPacket + protocolTag**。
### 9.2 CANopen规划
| 层级 | 计划 |
|------|------|
| **edge** | `CanIngress` + SocketCAN/`can0` 通道SDO/PDO 读对象字典 → `RawPacket``protocolTag: "canopen"` |
| **配置** | `physical_channels` 增加 `transport: can``interface: can0``bitrate: 500000``physical_devices` 增加 `nodeId`、对象索引 |
| **daemon** | 已有 `protocol_canopen.so`Framer/Parser 占位),绑定仍走 `bindings_global` |
| **EDS/DCF** | edge 侧加载 EDS 映射物理点 ↔ COB-ID/索引daemon 不接收 EDS 全量 |
### 9.3 以太网设备(规划)
| 类型 | edge 侧 | 说明 |
|------|---------|------|
| **Modbus TCP** | 已支持 | `physical_channels``transport: tcp``host` + `tcpPort` |
| **裸 TCP / 自定义帧** | `EthIngress`(规划) | 按 channel 配置帧格式或插件化 driver |
| **UDP 组播** | 待定 | 适用于部分工业相机/传感器 |
以太网通道示例Modbus TCP已可用
```json
{
"channelId": "eth_modbus_0",
"protocol": "modbus",
"transport": "tcp",
"host": "192.168.1.10",
"tcpPort": 502,
"pollIntervalMs": 200
}
```
### 9.4 edge 侧待增强Modbus 及通用)
| 项 | 说明 |
|----|------|
| **ModbusPollScheduler** | 合并相邻寄存器读,减少串口占用 |
| **按点 pollInterval** | 不同信号类型不同采集周期 |
| **变化上报 / deadband** | 物理层过滤,减少 uplink 流量 |
| **DeviceDiscovery** | 总线扫描、动态写入 `physical_devices` |
| **多 ingress 并存** | 同一 edge 上 Modbus + CANopen 并行 |
上述优化均在**物理域**完成,不改变 daemon 双域模型。
---
## 10. 约束CONSTRAINTS
- edge **不得**直接操作 OPC UA 或加载 `protocol_*.so` 解析插件
- edge **不得**加载 `logical_points.json` / `bindings_global.json`
- ingress 实现须遵循 `IIngressPort`,便于接入 CANopen/Ethernet driver
- 上行统一为 `RawPacket``sourceId` 必须是 **physicalPointId**(含 edgeId 前缀)
- daemon 不可达时 edge **必须**能独立启动并采集uplink 后台重连)
---
## 11. 相关文档
| 文档 | 内容 |
|------|------|
| [edge_device_setup.md](edge_device_setup.md) | 配置目录、双域、libmodbus、Mock |
| [UPLINK_WIRE_PROTOCOL.md](UPLINK_WIRE_PROTOCOL.md) | TCP JSON 线协议 |
| [softbus_daemon.md](softbus_daemon.md) | daemon 侧 Pipeline 与 OPC |
| [plugins.md](plugins.md) | protocol 插件daemon 解析用) |

View File

@@ -31,9 +31,8 @@ softbus::core::ParseResult ModbusParser::parseFrame(const uint8_t* data, std::si
result.message.value = {
{"protocol", "modbus"},
{"function", 0x03},
{"register", 40001},
{"registerValue", registerValue},
{"value", static_cast<double>(registerValue) * 0.1}
{"value", static_cast<double>(registerValue)}
};
return result;
}

View File

@@ -3,6 +3,7 @@
#include <softbus_api/ExecuteCommand.h>
#include <softbus_bus/Bus.h>
#include <softbus_command/CommandContext.h>
#include <softbus_registry/BindingRegistry.h>
#include <softbus_registry/PointRegistry.h>
#include <chrono>
@@ -17,6 +18,7 @@ namespace softbus::command {
class CommandService {
public:
explicit CommandService(softbus::registry::PointRegistry* registry,
softbus::registry::BindingRegistry* bindings = nullptr,
std::chrono::milliseconds timeout = std::chrono::milliseconds(5000));
void attachBus(softbus::bus::Bus* bus);
@@ -28,6 +30,7 @@ private:
std::string generateRequestId() const;
softbus::registry::PointRegistry* registry_{nullptr};
softbus::registry::BindingRegistry* bindings_{nullptr};
softbus::bus::Bus* bus_{nullptr};
softbus::bus::Bus::SubscriptionId ackSub_{0};
std::chrono::milliseconds timeout_;

View File

@@ -6,8 +6,10 @@
namespace softbus::command {
CommandService::CommandService(softbus::registry::PointRegistry* registry,
softbus::registry::BindingRegistry* bindings,
std::chrono::milliseconds timeout)
: registry_(registry)
, bindings_(bindings)
, timeout_(timeout)
{
}
@@ -84,6 +86,14 @@ CommandContext CommandService::submit(const softbus::api::ExecuteCommand& comman
softbus::bus::BusMessage msg;
msg.type = softbus::bus::BusPayloadType::ExecuteCommand;
msg.command = ctx.command;
if (bindings_) {
const auto logicalId = entry->deviceId + "/" + entry->pointId;
const auto route = bindings_->resolveRoute(logicalId);
if (route) {
msg.command.params["physicalPointId"] = route->physicalPointId;
msg.command.params["edgeId"] = route->edgeId;
}
}
bus_->publish(topic, msg);
{

View File

@@ -6,6 +6,7 @@ add_library(softbus_core STATIC
src/alignment/TimeAligner.cpp
src/plugin/PluginLoader.cpp
src/config/ProfileLoader.cpp
src/config/ConfigWatcher.cpp
src/platform/PlatformSignal.cpp
src/identity/DeviceIdentity.cpp
src/models/ObjectModel.cpp

View File

@@ -0,0 +1,23 @@
#pragma once
#include <string>
#include <vector>
namespace softbus::core {
class ConfigWatcher {
public:
void addWatch(const std::string& path);
void poll();
private:
struct WatchEntry {
std::string path;
long mtime{0};
};
std::vector<WatchEntry> watches_;
long readMtime(const std::string& path) const;
};
} // namespace softbus::core

View File

@@ -60,6 +60,7 @@ struct SoftbusEvent {
std::string eventType;
std::string message;
std::uint64_t timestampNs{0};
std::uint64_t lastHeartbeatNs{0};
nlohmann::json toJson() const;
static SoftbusEvent fromJson(const nlohmann::json& j);

View File

@@ -12,9 +12,13 @@ public:
static void install(Handler handler);
static bool shouldStop();
static void requestStop();
static bool reloadRequested();
static void requestReload();
static void clearReload();
private:
static std::atomic<bool> stopRequested_;
static std::atomic<bool> reloadRequested_;
};
} // namespace softbus::core

View File

@@ -0,0 +1,41 @@
#include <softbus_core/config/ConfigWatcher.h>
#include <softbus_core/platform/PlatformSignal.h>
#include <sys/stat.h>
namespace softbus::core {
void ConfigWatcher::addWatch(const std::string& path)
{
WatchEntry entry;
entry.path = path;
entry.mtime = readMtime(path);
watches_.push_back(std::move(entry));
}
long ConfigWatcher::readMtime(const std::string& path) const
{
struct stat st{};
if (stat(path.c_str(), &st) != 0) {
return 0;
}
return st.st_mtime;
}
void ConfigWatcher::poll()
{
for (auto& watch : watches_) {
const long mtime = readMtime(watch.path);
if (mtime != 0 && watch.mtime != 0 && mtime != watch.mtime) {
watch.mtime = mtime;
PlatformSignal::requestReload();
return;
}
if (watch.mtime == 0 && mtime != 0) {
watch.mtime = mtime;
}
}
}
} // namespace softbus::core

View File

@@ -68,12 +68,16 @@ SoftbusAck SoftbusAck::fromJson(const nlohmann::json& j)
nlohmann::json SoftbusEvent::toJson() const
{
return nlohmann::json{
nlohmann::json j{
{"deviceId", deviceId},
{"eventType", eventType},
{"message", message},
{"timestamp", timestampNs}
{"timestamp", timestampNs},
};
if (lastHeartbeatNs != 0) {
j["lastHeartbeat"] = lastHeartbeatNs;
}
return j;
}
SoftbusEvent SoftbusEvent::fromJson(const nlohmann::json& j)
@@ -83,6 +87,7 @@ SoftbusEvent SoftbusEvent::fromJson(const nlohmann::json& j)
ev.eventType = j.value("eventType", "");
ev.message = j.value("message", "");
ev.timestampNs = j.value("timestamp", static_cast<std::uint64_t>(0));
ev.lastHeartbeatNs = j.value("lastHeartbeat", static_cast<std::uint64_t>(0));
return ev;
}

View File

@@ -9,28 +9,31 @@
namespace softbus::core {
std::atomic<bool> PlatformSignal::stopRequested_{false};
std::atomic<bool> PlatformSignal::reloadRequested_{false};
#ifdef SOFTBUS_PLATFORM_WINDOWS
namespace {
BOOL WINAPI consoleHandler(DWORD signal)
{
if (signal == CTRL_C_EVENT || signal == CTRL_BREAK_EVENT || signal == CTRL_CLOSE_EVENT) {
PlatformSignal::requestStop();
return TRUE;
}
return FALSE;
BOOL WINAPI consoleHandler(DWORD signal)
{
if (signal == CTRL_C_EVENT || signal == CTRL_BREAK_EVENT || signal == CTRL_CLOSE_EVENT) {
PlatformSignal::requestStop();
return TRUE;
}
return FALSE;
}
} // namespace
#endif
void PlatformSignal::install(Handler handler)
{
stopRequested_.store(false);
reloadRequested_.store(false);
#ifdef SOFTBUS_PLATFORM_WINDOWS
SetConsoleCtrlHandler(consoleHandler, TRUE);
#else
std::signal(SIGINT, [](int) { requestStop(); });
std::signal(SIGTERM, [](int) { requestStop(); });
std::signal(SIGHUP, [](int) { requestReload(); });
#endif
(void)handler;
}
@@ -45,4 +48,19 @@ void PlatformSignal::requestStop()
stopRequested_.store(true);
}
bool PlatformSignal::reloadRequested()
{
return reloadRequested_.load();
}
void PlatformSignal::requestReload()
{
reloadRequested_.store(true);
}
void PlatformSignal::clearReload()
{
reloadRequested_.store(false);
}
} // namespace softbus::core

View File

@@ -69,10 +69,7 @@ std::string commandResultJson(const softbus::command::CommandContext& ctx)
} // namespace
CoreService::CoreService()
: commandService_(&registry_)
{
}
CoreService::CoreService() = default;
CoreService::~CoreService()
{
@@ -111,94 +108,114 @@ std::string CoreService::resolvePluginPath(const std::string& pluginName,
}
return fs::absolute(profileDir / "lib" / pluginFileNames(pluginName).front()).string();
}
// 主函数启动入口
// profilePath: 配置文件路径
// pointTablePath: 点表文件路径
// return: 是否启动成功
bool CoreService::start(const std::string& profilePath, const std::string& pointTablePath)
bool CoreService::loadLogicalConfig(const std::string& logicalPointsPath)
{
// 加载配置文件
const fs::path configDir = fs::path(logicalPointsPath).parent_path();
registry_.loadTopicRules((configDir / "topic_rule.json").string());
if (registry_.loadLogicalPoints(logicalPointsPath)) {
return true;
}
return registry_.loadPointTable(logicalPointsPath);
}
bool CoreService::start(const std::string& profilePath,
const std::string& logicalPointsPath,
const std::string& bindingsGlobalPath)
{
logicalPointsPath_ = logicalPointsPath;
bindingsGlobalPath_ = bindingsGlobalPath;
if (!profile_.loadFromFile(profilePath)) {
SB_LOG_ERROR("CoreService", "Failed to load profile: ", profilePath);
return false;
}
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);
if (!loadLogicalConfig(logicalPointsPath)) {
SB_LOG_ERROR("CoreService", "Failed to load logical points: ", logicalPointsPath);
return false;
}
model_ = registry_.tree();
if (!bindingsGlobalPath.empty()) {
if (!bindingRegistry_.loadGlobal(bindingsGlobalPath)) {
SB_LOG_ERROR("CoreService", "Failed to load bindings: ", bindingsGlobalPath);
return false;
}
configWatcher_.addWatch(bindingsGlobalPath);
}
configWatcher_.addWatch(logicalPointsPath);
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);
commandService_ = std::make_unique<softbus::command::CommandService>(
&registry_, &bindingRegistry_);
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<softbus::opcua::UaServerEngine>();
// 启动OPC UA服务
if (!uaEngine_->start(profile_.opcua.port, profile_.opcua.applicationName, security)) {
return false;
}
// 创建OPC UA模型绑定器
uaBinder_ = std::make_unique<softbus::opcua::UaModelBinder>(uaEngine_->nativeServer());
// 创建OPC UA方法绑定器
uaMethodBinder_ = std::make_unique<softbus::opcua::UaMethodBinder>(uaEngine_->nativeServer());
// 创建OPC UA事件桥接器
uaEventBridge_ = std::make_unique<softbus::opcua::UaEventBridge>(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<softbus::pipeline::PipelineEngine>(
&modbusPlugin_, uaBinder_.get(), &registry_);
&modbusPlugin_, uaBinder_.get(), &registry_, &bindingRegistry_);
pipeline_->attachBus(&bus);
// 创建TCP上行监听服务器
uplink_ = std::make_unique<softbus::transport::TcpUplinkServer>(
io_, profile_.uplink.listenHost, profile_.uplink.listenPort);
// 将消息总线绑定到TCP上行监听服务器
transportBridge_.attach(&bus, uplink_.get());
// 绑定OPC UA方法执行命令回调
// 将OPC UA方法执行命令回调绑定到消息总线
uaMethodBinder_->bindExecuteCommand([this](const softbus::api::ExecuteCommand& cmd) {
const auto ctx = commandService_.submit(cmd, true);
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 (bus+registry+command pipeline)");
SB_LOG_INFO("CoreService", "Daemon started (dual-domain+bindings)");
return true;
}
void CoreService::reloadConfig()
{
softbus::registry::PointRegistry previous = registry_;
if (!registry_.reloadPointTable(logicalPointsPath_)) {
SB_LOG_WARN("CoreService", "Logical points reload failed");
return;
}
model_ = registry_.tree();
if (!bindingsGlobalPath_.empty()) {
bindingRegistry_.reloadGlobal(bindingsGlobalPath_);
}
const auto added = registry_.diff(previous);
SB_LOG_INFO("CoreService", "Config reloaded, new logical points=", added.size());
}
void CoreService::stop()
{
if (!running_) {
@@ -227,6 +244,11 @@ void CoreService::run(bool selfTest)
const auto selfTestStart = std::chrono::steady_clock::now();
while (!softbus::core::PlatformSignal::shouldStop()) {
configWatcher_.poll();
if (softbus::core::PlatformSignal::reloadRequested()) {
reloadConfig();
softbus::core::PlatformSignal::clearReload();
}
if (uaEngine_) {
uaEngine_->iterate();
}
@@ -238,7 +260,7 @@ void CoreService::run(bool selfTest)
cmd.objectRef = "DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/ValveOpening";
cmd.command = "write";
cmd.params = {{"value", 60.0}};
const auto ctx = commandService_.submit(cmd, true);
const auto ctx = commandService_->submit(cmd, true);
SB_LOG_INFO("CoreService", "self-test command state=",
softbus::command::commandStateToString(ctx.state),
" executedValue=", ctx.executedValue);

View File

@@ -5,23 +5,27 @@
int main(int argc, char* argv[])
{
std::string profilePath = "config/daemon_profile.json";
std::string pointTablePath = "config/point_table.json";
std::string profilePath = "config/daemon/daemon_profile.json";
std::string logicalPointsPath = "config/daemon/logical_points.json";
std::string bindingsGlobalPath = "config/daemon/bindings_global.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 == "--point-table" || arg == "--model") && i + 1 < argc) {
pointTablePath = argv[++i];
} else if ((arg == "--logical-points" || arg == "--point-table" || arg == "--model")
&& i + 1 < argc) {
logicalPointsPath = argv[++i];
} else if (arg == "--bindings-global" && i + 1 < argc) {
bindingsGlobalPath = argv[++i];
} else if (arg == "--self-test") {
selfTest = true;
}
}
softbus::daemon::CoreService service;
if (!service.start(profilePath, pointTablePath)) {
if (!service.start(profilePath, logicalPointsPath, bindingsGlobalPath)) {
std::cerr << "Failed to start softbus_daemon" << std::endl;
return 1;
}

View File

@@ -1,12 +1,18 @@
# role: edge 用于采集物理设备数据,并将其转换为软总线数据模型
softbus_package(softbus_edge)
add_executable(softbus_edge
src/main.cpp
src/MockModbusIngress.cpp
src/ModbusIngress.cpp
src/ModbusChannel.cpp
src/ModbusEgress.cpp
src/PhysicalRegistry.cpp
src/EdgeConfig.cpp
src/EdgeRuntime.cpp
src/UplinkClient.cpp
src/DeviceFactory.cpp
src/DeviceDiscovery.cpp
src/EdgeConfig.cpp
src/ShadowChannelStore.cpp
src/CommandExecutor.cpp
)
@@ -20,4 +26,6 @@ target_link_libraries(softbus_edge PRIVATE
softbus_api
asio
)
# 添加 libmodbus 库
softbus_apply_libmodbus(softbus_edge)
softbus_apply_platform_libs(softbus_edge)

View File

@@ -6,11 +6,16 @@
namespace softbus::edge {
struct EdgeConfig {
std::string edgeId{"edge01"};
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"};
int uplinkReconnectPeriodMs{5000};
std::string physicalChannelsPath{"physical_channels.json"};
std::string physicalDevicesPath{"physical_devices.json"};
std::string bindingsLocalPath{"bindings_local.json"};
std::string bindingsPatchPath;
bool loadFromFile(const std::string& path);
};

View File

@@ -0,0 +1,42 @@
#pragma once
#include <softbus_edge/config/PhysicalTypes.h>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
namespace softbus::edge {
struct ResolvedPhysicalPoint {
PhysicalPointDef point;
PhysicalDeviceDef device;
PhysicalChannelDef channel;
};
class PhysicalRegistry {
public:
bool loadChannels(const std::string& path);
bool loadDevices(const std::string& path, const std::string& edgeIdPrefix = {});
bool reloadChannels(const std::string& path);
bool reloadDevices(const std::string& path, const std::string& edgeIdPrefix = {});
const std::vector<PhysicalChannelDef>& channels() const { return channels_; }
const std::vector<PhysicalDeviceDef>& devices() const { return devices_; }
const std::vector<PhysicalPointDef>& pollPoints() const { return pollPoints_; }
std::optional<ResolvedPhysicalPoint> resolvePoint(const std::string& physicalPointId) const;
std::optional<PhysicalChannelDef> findChannel(const std::string& channelId) const;
private:
void rebuildPollIndex(const std::string& edgeIdPrefix);
std::vector<PhysicalChannelDef> channels_;
std::vector<PhysicalDeviceDef> devices_;
std::vector<PhysicalPointDef> pollPoints_;
std::unordered_map<std::string, std::size_t> channelIndex_;
std::unordered_map<std::string, ResolvedPhysicalPoint> pointIndex_;
};
} // namespace softbus::edge

View File

@@ -0,0 +1,41 @@
#pragma once
#include <cstdint>
#include <string>
#include <vector>
namespace softbus::edge {
struct PhysicalChannelDef {
std::string channelId;
std::string protocol{"modbus"};
std::string transport{"rtu"};
std::string port;
std::string host;
uint16_t tcpPort{502};
int baudRate{9600};
char parity{'N'};
int dataBits{8};
int stopBits{1};
int pollIntervalMs{500};
};
struct PhysicalPointDef {
std::string physicalPointId;
int registerAddress{0};
int functionCode{3};
int writeFunctionCode{6};
double scale{1.0};
std::string signalType{"AI"};
bool writable{false};
};
struct PhysicalDeviceDef {
std::string physicalDeviceId;
std::string channelId;
std::string protocol{"modbus"};
int slaveId{1};
std::vector<PhysicalPointDef> points;
};
} // namespace softbus::edge

View File

@@ -0,0 +1,32 @@
#pragma once
#include <softbus_edge/config/PhysicalRegistry.h>
#include <cstdint>
#include <optional>
#include <string>
#include <unordered_map>
namespace softbus::edge {
class ModbusChannelManager {
public:
bool openChannel(const PhysicalChannelDef& channel);
void closeChannel(const std::string& channelId);
void closeAll();
std::optional<uint16_t> readRegister(const ResolvedPhysicalPoint& point);
bool writeRegister(const ResolvedPhysicalPoint& point, uint16_t rawValue);
private:
struct ChannelHandle {
PhysicalChannelDef config;
#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS
void* ctx{nullptr};
#endif
};
std::unordered_map<std::string, ChannelHandle> channels_;
int modbusAddress(int registerAddress) const;
};
} // namespace softbus::edge

View File

@@ -1,26 +1,38 @@
#pragma once
#include <softbus_api/ExecuteCommand.h>
#include <softbus_edge/config/PhysicalRegistry.h>
#include <softbus_edge/egress/ModbusEgress.h>
#include <softbus_edge/egress/ShadowChannelStore.h>
#include <softbus_registry/PointRegistry.h>
#include <softbus_edge/egress/UplinkClient.h>
#include <softbus_registry/BindingRegistry.h>
namespace softbus::edge {
class CommandExecutor {
public:
CommandExecutor(softbus::registry::PointRegistry* registry,
ShadowChannelStore* store,
CommandExecutor(PhysicalRegistry* physicalRegistry,
ModbusEgress* modbusEgress,
ShadowChannelStore* mockStore,
softbus::registry::BindingRegistry* localBindings,
UplinkClient* uplink,
std::string deviceId);
std::string deviceId,
bool useMock = false);
softbus::core::SoftbusAck execute(const softbus::api::ExecuteCommand& command);
private:
softbus::registry::PointRegistry* registry_{nullptr};
ShadowChannelStore* store_{nullptr};
void fillAckIdentity(softbus::core::SoftbusAck& ack,
const softbus::api::ExecuteCommand& command,
const std::string& physicalPointId) const;
PhysicalRegistry* physicalRegistry_{nullptr};
ModbusEgress* modbusEgress_{nullptr};
ShadowChannelStore* mockStore_{nullptr};
softbus::registry::BindingRegistry* localBindings_{nullptr};
UplinkClient* uplink_{nullptr};
std::string deviceId_;
bool useMock_{false};
};
} // namespace softbus::edge

View File

@@ -0,0 +1,23 @@
#pragma once
#include <softbus_edge/config/PhysicalRegistry.h>
#include <softbus_edge/driver/ModbusChannel.h>
#include <optional>
#include <string>
namespace softbus::edge {
class ModbusEgress {
public:
explicit ModbusEgress(PhysicalRegistry* registry, ModbusChannelManager* channels);
bool writeByPhysicalPointId(const std::string& physicalPointId, double engineeringValue,
int* rawOut = nullptr);
private:
PhysicalRegistry* registry_{nullptr};
ModbusChannelManager* channels_{nullptr};
};
} // namespace softbus::edge

View File

@@ -1,7 +1,5 @@
#pragma once
#include <softbus_registry/PointTableEntry.h>
#include <optional>
#include <string>
#include <unordered_map>
@@ -10,12 +8,11 @@ namespace softbus::edge {
class ShadowChannelStore {
public:
bool write(const softbus::registry::PointTableEntry& entry, double engineeringValue);
std::optional<double> read(const std::string& physicalChannel) const;
int rawRegisterValue(const softbus::registry::PointTableEntry& entry, double engineeringValue) const;
bool write(const std::string& physicalPointId, double engineeringValue);
std::optional<double> read(const std::string& physicalPointId) const;
private:
std::unordered_map<std::string, double> channels_;
std::unordered_map<std::string, double> points_;
};
} // namespace softbus::edge

View File

@@ -18,9 +18,11 @@ public:
bool connect();
void disconnect();
bool isConnected() const;
bool sendRawPacket(const softbus::core::RawPacket& packet);
bool sendAck(const softbus::core::SoftbusAck& ack);
bool sendHeartbeat(const std::string& deviceId);
bool sendEdgeRegister(const std::string& edgeId);
void setCommandHandler(std::function<void(const softbus::api::ExecuteCommand&)> handler);
void poll();

View File

@@ -1,5 +1,14 @@
#pragma once
#include <softbus_edge/config/EdgeConfig.h>
#include <softbus_edge/config/PhysicalRegistry.h>
#include <softbus_edge/driver/ModbusChannel.h>
#include <softbus_edge/egress/ModbusEgress.h>
#include <softbus_edge/ingress/IIngressPort.h>
#include <softbus_edge/ingress/MockModbusIngress.h>
#include <softbus_edge/ingress/ModbusIngress.h>
#include <memory>
#include <string>
namespace softbus::edge {
@@ -7,6 +16,12 @@ namespace softbus::edge {
class DeviceFactory {
public:
std::string createRuntimeDeviceId();
std::unique_ptr<IIngressPort> createIngress(const EdgeConfig& config,
PhysicalRegistry* registry,
ModbusChannelManager* channels,
bool useMock,
const std::string& mockObjectRef);
};
} // namespace softbus::edge

View File

@@ -0,0 +1,30 @@
#pragma once
#include <softbus_edge/config/PhysicalRegistry.h>
#include <softbus_edge/driver/ModbusChannel.h>
#include <softbus_edge/ingress/IIngressPort.h>
#include <chrono>
#include <string>
namespace softbus::edge {
class ModbusIngress : public IIngressPort {
public:
ModbusIngress(PhysicalRegistry* registry, ModbusChannelManager* channels, std::string edgeIdPrefix);
bool open() override;
void close() override;
std::vector<softbus::core::RawPacket> poll() override;
private:
std::vector<uint8_t> buildModbusFrame(int slaveId, uint16_t registerValue) const;
PhysicalRegistry* registry_{nullptr};
ModbusChannelManager* channels_{nullptr};
std::string edgeIdPrefix_;
bool open_{false};
std::chrono::steady_clock::time_point lastPoll_{};
};
} // namespace softbus::edge

View File

@@ -0,0 +1,51 @@
#pragma once
#include <softbus_edge/config/EdgeConfig.h>
#include <softbus_edge/config/PhysicalRegistry.h>
#include <softbus_edge/driver/ModbusChannel.h>
#include <softbus_edge/egress/CommandExecutor.h>
#include <softbus_edge/egress/ModbusEgress.h>
#include <softbus_edge/egress/ShadowChannelStore.h>
#include <softbus_edge/egress/UplinkClient.h>
#include <softbus_edge/factory/DeviceFactory.h>
#include <softbus_edge/ingress/IIngressPort.h>
#include <softbus_registry/BindingRegistry.h>
#include <chrono>
#include <memory>
#include <string>
namespace softbus::edge {
class EdgeRuntime {
public:
bool init(const EdgeConfig& config, bool useMock, const std::string& mockObjectRef);
void reloadPhysical();
void reloadBindingsPatch();
void shutdown();
void tick();
UplinkClient* uplink() { return uplink_.get(); }
bool isUplinkConnected() const;
private:
void tickUplink();
void pollIngress();
EdgeConfig config_;
bool useMock_{false};
std::string mockObjectRef_;
softbus::registry::BindingRegistry localBindings_;
PhysicalRegistry physicalRegistry_;
ModbusChannelManager modbusChannels_;
ModbusEgress modbusEgress_{&physicalRegistry_, &modbusChannels_};
ShadowChannelStore mockStore_;
std::unique_ptr<UplinkClient> uplink_;
std::unique_ptr<IIngressPort> ingress_;
std::unique_ptr<CommandExecutor> executor_;
DeviceFactory factory_;
std::chrono::steady_clock::time_point lastUplinkReconnectAttempt_{};
};
} // namespace softbus::edge

View File

@@ -1,42 +1,49 @@
#include <softbus_edge/egress/CommandExecutor.h>
#include <softbus_core/identity/DeviceIdentity.h>
#include <softbus_core/logging/Logger.h>
namespace softbus::edge {
CommandExecutor::CommandExecutor(softbus::registry::PointRegistry* registry,
ShadowChannelStore* store,
CommandExecutor::CommandExecutor(PhysicalRegistry* physicalRegistry,
ModbusEgress* modbusEgress,
ShadowChannelStore* mockStore,
softbus::registry::BindingRegistry* localBindings,
UplinkClient* uplink,
std::string deviceId)
: registry_(registry)
, store_(store)
std::string deviceId,
bool useMock)
: physicalRegistry_(physicalRegistry)
, modbusEgress_(modbusEgress)
, mockStore_(mockStore)
, localBindings_(localBindings)
, uplink_(uplink)
, deviceId_(std::move(deviceId))
, useMock_(useMock)
{
}
void CommandExecutor::fillAckIdentity(softbus::core::SoftbusAck& ack,
const softbus::api::ExecuteCommand& command,
const std::string& physicalPointId) const
{
if (!command.objectRef.empty()) {
const auto ref = softbus::core::ObjectRef::fromPath(command.objectRef);
if (ref.isValid() && ref.hasPoint()) {
ack.deviceId = ref.device;
ack.pointId = ref.point;
return;
}
}
ack.deviceId = deviceId_;
ack.pointId = physicalPointId;
}
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";
@@ -44,21 +51,66 @@ softbus::core::SoftbusAck CommandExecutor::execute(const softbus::api::ExecuteCo
}
const double value = command.params.value("value", 0.0);
std::string error;
if (!registry_->validateWriteValue(*entry, value, &error)) {
std::string physicalPointId = command.params.value("physicalPointId", "");
if (physicalPointId.empty() && localBindings_) {
const auto logicalId = command.params.value("logicalPointId", "");
if (!logicalId.empty()) {
const auto resolved = localBindings_->resolvePhysicalLocal(logicalId);
if (resolved) {
physicalPointId = *resolved;
}
}
}
if (physicalPointId.empty()) {
ack.success = false;
ack.error = error;
ack.error = "no physicalPointId resolved (edge does not load logical point table)";
return ack;
}
store_->write(*entry, value);
if (useMock_ && mockStore_) {
mockStore_->write(physicalPointId, value);
fillAckIdentity(ack, command, physicalPointId);
ack.executedValue = value;
ack.success = true;
if (uplink_) {
uplink_->sendAck(ack);
}
return ack;
}
if (!modbusEgress_ || !physicalRegistry_) {
ack.success = false;
ack.error = "modbus egress not initialized";
return ack;
}
const auto resolved = physicalRegistry_->resolvePoint(physicalPointId);
if (!resolved) {
ack.success = false;
ack.error = "unknown physicalPointId";
return ack;
}
if (!resolved->point.writable) {
ack.success = false;
ack.error = "physical point is not writable";
return ack;
}
int raw = 0;
if (!modbusEgress_->writeByPhysicalPointId(physicalPointId, value, &raw)) {
ack.success = false;
ack.error = "modbus write failed";
return ack;
}
fillAckIdentity(ack, command, physicalPointId);
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)");
"write ", physicalPointId, " value=", value, " raw=", raw);
if (uplink_) {
uplink_->sendAck(ack);

View File

@@ -9,4 +9,16 @@ std::string DeviceFactory::createRuntimeDeviceId()
return softbus::core::DeviceIdentity::generateRuntimeDeviceId();
}
std::unique_ptr<IIngressPort> DeviceFactory::createIngress(const EdgeConfig& config,
PhysicalRegistry* registry,
ModbusChannelManager* channels,
bool useMock,
const std::string& mockObjectRef)
{
if (useMock) {
return std::make_unique<MockModbusIngress>(mockObjectRef, config.edgeId);
}
return std::make_unique<ModbusIngress>(registry, channels, config.edgeId);
}
} // namespace softbus::edge

View File

@@ -1,10 +1,28 @@
#include <softbus_edge/config/EdgeConfig.h>
#include <filesystem>
#include <fstream>
#include <nlohmann/json.hpp>
namespace softbus::edge {
namespace fs = std::filesystem;
namespace {
void resolveConfigPath(const fs::path& baseDir, std::string& path)
{
if (path.empty()) {
return;
}
const fs::path candidate = baseDir / path;
if (fs::exists(candidate)) {
path = fs::absolute(candidate).string();
}
}
} // namespace
bool EdgeConfig::loadFromFile(const std::string& path)
{
std::ifstream ifs(path);
@@ -17,11 +35,22 @@ bool EdgeConfig::loadFromFile(const std::string& path)
return false;
}
const auto& edge = j["edge"];
edgeId = edge.value("edgeId", edgeId);
deviceId = edge.value("deviceId", deviceId);
daemonHost = edge.value("daemonHost", daemonHost);
daemonPort = static_cast<uint16_t>(edge.value("daemonPort", static_cast<int>(daemonPort)));
heartbeatPeriodMs = edge.value("heartbeatPeriodMs", heartbeatPeriodMs);
pointTablePath = edge.value("pointTable", pointTablePath);
uplinkReconnectPeriodMs = edge.value("uplinkReconnectPeriodMs", uplinkReconnectPeriodMs);
physicalChannelsPath = edge.value("physicalChannels", physicalChannelsPath);
physicalDevicesPath = edge.value("physicalDevices", physicalDevicesPath);
bindingsLocalPath = edge.value("bindingsLocal", bindingsLocalPath);
bindingsPatchPath = edge.value("bindingsPatch", bindingsPatchPath);
const fs::path baseDir = fs::path(path).parent_path();
resolveConfigPath(baseDir, physicalChannelsPath);
resolveConfigPath(baseDir, physicalDevicesPath);
resolveConfigPath(baseDir, bindingsLocalPath);
resolveConfigPath(baseDir, bindingsPatchPath);
return true;
}

View File

@@ -0,0 +1,138 @@
#include <softbus_edge/runtime/EdgeRuntime.h>
#include <softbus_core/logging/Logger.h>
namespace softbus::edge {
bool EdgeRuntime::init(const EdgeConfig& config, bool useMock, const std::string& mockObjectRef)
{
config_ = config;
useMock_ = useMock;
mockObjectRef_ = mockObjectRef;
if (!useMock) {
if (!physicalRegistry_.loadChannels(config.physicalChannelsPath)) {
SB_LOG_ERROR("EdgeRuntime", "Failed to load physical channels");
return false;
}
if (!physicalRegistry_.loadDevices(config.physicalDevicesPath, config.edgeId)) {
SB_LOG_ERROR("EdgeRuntime", "Failed to load physical devices");
return false;
}
}
if (!config.bindingsLocalPath.empty()) {
localBindings_.loadLocal(config.bindingsLocalPath, config.edgeId);
}
uplink_ = std::make_unique<UplinkClient>(config.daemonHost, config.daemonPort);
executor_ = std::make_unique<CommandExecutor>(
&physicalRegistry_, &modbusEgress_, &mockStore_, &localBindings_,
uplink_.get(), config.deviceId, useMock);
uplink_->setCommandHandler([this](const softbus::api::ExecuteCommand& cmd) {
if (executor_) {
executor_->execute(cmd);
}
});
ingress_ = factory_.createIngress(config, &physicalRegistry_, &modbusChannels_, useMock, mockObjectRef);
if (!ingress_->open()) {
SB_LOG_ERROR("EdgeRuntime", "Failed to open ingress");
return false;
}
if (!uplink_->connect()) {
SB_LOG_WARN(
"EdgeRuntime",
"Daemon unreachable at ",
config.daemonHost,
":",
config.daemonPort,
"; running standalone, uplink will retry in background");
} else {
uplink_->sendEdgeRegister(config.edgeId);
}
lastUplinkReconnectAttempt_ = std::chrono::steady_clock::now();
return true;
}
void EdgeRuntime::reloadPhysical()
{
if (useMock_) {
return;
}
ingress_->close();
modbusChannels_.closeAll();
physicalRegistry_.reloadChannels(config_.physicalChannelsPath);
physicalRegistry_.reloadDevices(config_.physicalDevicesPath, config_.edgeId);
ingress_->open();
SB_LOG_INFO("EdgeRuntime", "Physical config reloaded");
}
void EdgeRuntime::reloadBindingsPatch()
{
if (config_.bindingsPatchPath.empty()) {
return;
}
localBindings_.applyPatch(config_.bindingsPatchPath);
SB_LOG_INFO("EdgeRuntime", "Bindings patch applied");
}
void EdgeRuntime::shutdown()
{
if (ingress_) {
ingress_->close();
}
modbusChannels_.closeAll();
if (uplink_) {
uplink_->disconnect();
}
}
bool EdgeRuntime::isUplinkConnected() const
{
return uplink_ && uplink_->isConnected();
}
void EdgeRuntime::tickUplink()
{
if (!uplink_) {
return;
}
uplink_->poll();
if (uplink_->isConnected()) {
return;
}
const auto now = std::chrono::steady_clock::now();
if (now - lastUplinkReconnectAttempt_ <
std::chrono::milliseconds(config_.uplinkReconnectPeriodMs)) {
return;
}
lastUplinkReconnectAttempt_ = now;
if (uplink_->connect()) {
uplink_->sendEdgeRegister(config_.edgeId);
SB_LOG_INFO("EdgeRuntime", "Connected to daemon at ", config_.daemonHost, ":", config_.daemonPort);
}
}
void EdgeRuntime::pollIngress()
{
if (!ingress_) {
return;
}
for (const auto& packet : ingress_->poll()) {
if (uplink_ && uplink_->isConnected()) {
uplink_->sendRawPacket(packet);
}
}
}
void EdgeRuntime::tick()
{
tickUplink();
pollIngress();
}
} // namespace softbus::edge

View File

@@ -0,0 +1,123 @@
#include <softbus_edge/driver/ModbusChannel.h>
#include <softbus_core/logging/Logger.h>
#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS
#include <modbus.h>
#endif
#include <cmath>
namespace softbus::edge {
int ModbusChannelManager::modbusAddress(int registerAddress) const
{
if (registerAddress >= 40001) {
return registerAddress - 40001;
}
if (registerAddress >= 30001) {
return registerAddress - 30001;
}
return registerAddress;
}
bool ModbusChannelManager::openChannel(const PhysicalChannelDef& channel)
{
closeChannel(channel.channelId);
ChannelHandle handle;
handle.config = channel;
#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS
if (channel.transport == "tcp") {
handle.ctx = modbus_new_tcp(channel.host.c_str(), channel.tcpPort);
} else {
char parity = channel.parity;
handle.ctx = modbus_new_rtu(
channel.port.c_str(), channel.baudRate, parity, channel.dataBits, channel.stopBits);
}
if (!handle.ctx) {
SB_LOG_ERROR("ModbusChannel", "Failed to create modbus context for ", channel.channelId);
return false;
}
modbus_set_response_timeout(static_cast<modbus_t*>(handle.ctx), 0, 500000);
if (modbus_connect(static_cast<modbus_t*>(handle.ctx)) != 0) {
SB_LOG_ERROR("ModbusChannel", "Failed to connect channel ", channel.channelId);
modbus_free(static_cast<modbus_t*>(handle.ctx));
return false;
}
channels_[channel.channelId] = handle;
SB_LOG_INFO("ModbusChannel", "Opened channel ", channel.channelId);
return true;
#else
(void)handle;
channels_[channel.channelId] = ChannelHandle{channel};
SB_LOG_WARN("ModbusChannel", "libmodbus unavailable; channel ", channel.channelId, " simulated");
return true;
#endif
}
void ModbusChannelManager::closeChannel(const std::string& channelId)
{
const auto it = channels_.find(channelId);
if (it == channels_.end()) {
return;
}
#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS
if (it->second.ctx) {
modbus_close(static_cast<modbus_t*>(it->second.ctx));
modbus_free(static_cast<modbus_t*>(it->second.ctx));
}
#endif
channels_.erase(it);
}
void ModbusChannelManager::closeAll()
{
for (const auto& [id, _] : channels_) {
(void)_;
closeChannel(id);
}
}
std::optional<uint16_t> ModbusChannelManager::readRegister(const ResolvedPhysicalPoint& point)
{
#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS
const auto it = channels_.find(point.channel.channelId);
if (it == channels_.end() || !it->second.ctx) {
return std::nullopt;
}
modbus_set_slave(static_cast<modbus_t*>(it->second.ctx), point.device.slaveId);
uint16_t value = 0;
const int addr = modbusAddress(point.point.registerAddress);
const int rc = modbus_read_registers(
static_cast<modbus_t*>(it->second.ctx), addr, 1, &value);
if (rc != 1) {
return std::nullopt;
}
return value;
#else
static uint16_t sim = 250;
sim = static_cast<uint16_t>((sim + 1) % 500);
(void)point;
return sim;
#endif
}
bool ModbusChannelManager::writeRegister(const ResolvedPhysicalPoint& point, uint16_t rawValue)
{
#if defined(SOFTBUS_HAS_LIBMODBUS) && SOFTBUS_HAS_LIBMODBUS
const auto it = channels_.find(point.channel.channelId);
if (it == channels_.end() || !it->second.ctx) {
return false;
}
modbus_set_slave(static_cast<modbus_t*>(it->second.ctx), point.device.slaveId);
const int addr = modbusAddress(point.point.registerAddress);
return modbus_write_register(static_cast<modbus_t*>(it->second.ctx), addr, rawValue) == 1;
#else
(void)point;
(void)rawValue;
return true;
#endif
}
} // namespace softbus::edge

View File

@@ -0,0 +1,33 @@
#include <softbus_edge/egress/ModbusEgress.h>
#include <cmath>
namespace softbus::edge {
ModbusEgress::ModbusEgress(PhysicalRegistry* registry, ModbusChannelManager* channels)
: registry_(registry)
, channels_(channels)
{
}
bool ModbusEgress::writeByPhysicalPointId(const std::string& physicalPointId,
double engineeringValue,
int* rawOut)
{
if (!registry_ || !channels_) {
return false;
}
const auto resolved = registry_->resolvePoint(physicalPointId);
if (!resolved || !resolved->point.writable) {
return false;
}
const int raw = resolved->point.scale <= 0.0
? static_cast<int>(std::lround(engineeringValue))
: static_cast<int>(std::lround(engineeringValue / resolved->point.scale));
if (rawOut) {
*rawOut = raw;
}
return channels_->writeRegister(*resolved, static_cast<uint16_t>(raw));
}
} // namespace softbus::edge

View File

@@ -0,0 +1,92 @@
#include <softbus_edge/ingress/ModbusIngress.h>
#include <softbus_core/identity/DeviceIdentity.h>
#include <softbus_core/time/HwClock.h>
namespace softbus::edge {
ModbusIngress::ModbusIngress(PhysicalRegistry* registry,
ModbusChannelManager* channels,
std::string edgeIdPrefix)
: registry_(registry)
, channels_(channels)
, edgeIdPrefix_(std::move(edgeIdPrefix))
{
}
bool ModbusIngress::open()
{
if (!registry_ || !channels_) {
return false;
}
for (const auto& ch : registry_->channels()) {
if (!channels_->openChannel(ch)) {
return false;
}
}
open_ = true;
lastPoll_ = std::chrono::steady_clock::now();
return true;
}
void ModbusIngress::close()
{
if (channels_) {
channels_->closeAll();
}
open_ = false;
}
std::vector<uint8_t> ModbusIngress::buildModbusFrame(int slaveId, uint16_t registerValue) const
{
return {
static_cast<uint8_t>(slaveId),
0x03,
0x02,
static_cast<uint8_t>((registerValue >> 8) & 0xFF),
static_cast<uint8_t>(registerValue & 0xFF),
0x00,
0x00
};
}
std::vector<softbus::core::RawPacket> ModbusIngress::poll()
{
if (!open_ || !registry_ || !channels_) {
return {};
}
const auto now = std::chrono::steady_clock::now();
int pollMs = 500;
if (!registry_->channels().empty()) {
pollMs = registry_->channels().front().pollIntervalMs;
}
if (now - lastPoll_ < std::chrono::milliseconds(pollMs)) {
return {};
}
lastPoll_ = now;
std::vector<softbus::core::RawPacket> packets;
for (const auto& pointDef : registry_->pollPoints()) {
const auto resolved = registry_->resolvePoint(pointDef.physicalPointId);
if (!resolved) {
continue;
}
const auto raw = channels_->readRegister(*resolved);
if (!raw) {
continue;
}
softbus::core::RawPacket packet;
packet.traceId = softbus::core::DeviceIdentity::generateTraceId();
packet.timestampNs = softbus::core::HwClock::nowNs();
packet.runtimeDeviceId = edgeIdPrefix_;
packet.sourceId = pointDef.physicalPointId;
packet.protocolTag = "modbus";
packet.payload = buildModbusFrame(resolved->device.slaveId, *raw);
packets.push_back(std::move(packet));
}
return packets;
}
} // namespace softbus::edge

View File

@@ -0,0 +1,152 @@
#include <softbus_edge/config/PhysicalRegistry.h>
#include <fstream>
#include <nlohmann/json.hpp>
namespace softbus::edge {
namespace {
std::string qualifyPointId(const std::string& edgeIdPrefix, const std::string& pointId)
{
if (edgeIdPrefix.empty()) {
return pointId;
}
if (pointId.find('/') != std::string::npos) {
return pointId;
}
return edgeIdPrefix + "/" + pointId;
}
} // namespace
bool PhysicalRegistry::loadChannels(const std::string& path)
{
std::ifstream ifs(path);
if (!ifs) {
return false;
}
nlohmann::json j;
ifs >> j;
channels_.clear();
channelIndex_.clear();
if (!j.contains("channels")) {
return true;
}
for (const auto& ch : j["channels"]) {
PhysicalChannelDef def;
def.channelId = ch.value("channelId", "");
def.protocol = ch.value("protocol", "modbus");
def.transport = ch.value("transport", "rtu");
def.port = ch.value("port", "");
def.host = ch.value("host", "");
def.tcpPort = static_cast<uint16_t>(ch.value("tcpPort", 502));
def.baudRate = ch.value("baudRate", 9600);
const std::string parity = ch.value("parity", "N");
def.parity = parity.empty() ? 'N' : parity[0];
def.dataBits = ch.value("dataBits", 8);
def.stopBits = ch.value("stopBits", 1);
def.pollIntervalMs = ch.value("pollIntervalMs", 500);
if (def.channelId.empty()) {
continue;
}
channelIndex_[def.channelId] = channels_.size();
channels_.push_back(std::move(def));
}
return true;
}
bool PhysicalRegistry::loadDevices(const std::string& path, const std::string& edgeIdPrefix)
{
std::ifstream ifs(path);
if (!ifs) {
return false;
}
nlohmann::json j;
ifs >> j;
devices_.clear();
pointIndex_.clear();
pollPoints_.clear();
if (!j.contains("devices")) {
return true;
}
for (const auto& dev : j["devices"]) {
PhysicalDeviceDef device;
device.physicalDeviceId = dev.value("physicalDeviceId", "");
device.channelId = dev.value("channelId", "");
device.protocol = dev.value("protocol", "modbus");
device.slaveId = dev.value("slaveId", 1);
if (!dev.contains("points")) {
devices_.push_back(std::move(device));
continue;
}
for (const auto& pt : dev["points"]) {
PhysicalPointDef point;
point.physicalPointId = qualifyPointId(
edgeIdPrefix, pt.value("physicalPointId", ""));
point.registerAddress = pt.value("register", 0);
point.functionCode = pt.value("function", 3);
point.writeFunctionCode = pt.value("writeFunction", 6);
point.scale = pt.value("scale", 1.0);
point.signalType = pt.value("signalType", "AI");
point.writable = pt.value("writable", false);
if (point.physicalPointId.empty() || point.registerAddress == 0) {
continue;
}
device.points.push_back(point);
}
devices_.push_back(std::move(device));
}
rebuildPollIndex(edgeIdPrefix);
return true;
}
bool PhysicalRegistry::reloadChannels(const std::string& path)
{
return loadChannels(path);
}
bool PhysicalRegistry::reloadDevices(const std::string& path, const std::string& edgeIdPrefix)
{
return loadDevices(path, edgeIdPrefix);
}
void PhysicalRegistry::rebuildPollIndex(const std::string& edgeIdPrefix)
{
(void)edgeIdPrefix;
pollPoints_.clear();
pointIndex_.clear();
for (const auto& device : devices_) {
const auto chIt = channelIndex_.find(device.channelId);
if (chIt == channelIndex_.end()) {
continue;
}
const auto& channel = channels_[chIt->second];
for (const auto& point : device.points) {
ResolvedPhysicalPoint resolved{point, device, channel};
pointIndex_[point.physicalPointId] = resolved;
pollPoints_.push_back(point);
}
}
}
std::optional<ResolvedPhysicalPoint> PhysicalRegistry::resolvePoint(
const std::string& physicalPointId) const
{
const auto it = pointIndex_.find(physicalPointId);
if (it == pointIndex_.end()) {
return std::nullopt;
}
return it->second;
}
std::optional<PhysicalChannelDef> PhysicalRegistry::findChannel(const std::string& channelId) const
{
const auto it = channelIndex_.find(channelId);
if (it == channelIndex_.end()) {
return std::nullopt;
}
return channels_[it->second];
}
} // namespace softbus::edge

View File

@@ -1,32 +1,20 @@
#include <softbus_edge/egress/ShadowChannelStore.h>
#include <cmath>
namespace softbus::edge {
bool ShadowChannelStore::write(const softbus::registry::PointTableEntry& entry,
double engineeringValue)
bool ShadowChannelStore::write(const std::string& physicalPointId, double engineeringValue)
{
channels_[entry.physicalChannel] = engineeringValue;
points_[physicalPointId] = engineeringValue;
return true;
}
std::optional<double> ShadowChannelStore::read(const std::string& physicalChannel) const
std::optional<double> ShadowChannelStore::read(const std::string& physicalPointId) const
{
const auto it = channels_.find(physicalChannel);
if (it == channels_.end()) {
const auto it = points_.find(physicalPointId);
if (it == points_.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<int>(std::lround(engineeringValue));
}
return static_cast<int>(std::lround(engineeringValue / entry.scale));
}
} // namespace softbus::edge

View File

@@ -22,6 +22,11 @@ void UplinkClient::disconnect()
client_->disconnect();
}
bool UplinkClient::isConnected() const
{
return client_->isConnected();
}
bool UplinkClient::sendRawPacket(const softbus::core::RawPacket& packet)
{
return client_->sendRawPacket(packet);
@@ -37,6 +42,11 @@ bool UplinkClient::sendHeartbeat(const std::string& deviceId)
return client_->sendHeartbeat(deviceId);
}
bool UplinkClient::sendEdgeRegister(const std::string& edgeId)
{
return client_->sendEdgeRegister(edgeId);
}
void UplinkClient::setCommandHandler(std::function<void(const softbus::api::ExecuteCommand&)> handler)
{
client_->setCommandHandler(std::move(handler));

View File

@@ -1,17 +1,17 @@
// role: edge 用于采集物理设备数据,并将其转换为软总线数据模型
// @TODO 1合并读以后建议在 edge 加物理层的差异化采集/变化上报,而不是加逻辑 binding 初筛。
// 计划里的 ModbusPollScheduler合并读还没单独实现这是 edge 侧最自然的下一步之一。
// CANopenedge 侧 CanIngress + SocketCANdaemon 侧已有 protocol_canopen 占位
// 以太网Modbus TCP 已支持transport: tcp裸 TCP / UDP 为规划项
// 待增强PollScheduler、按点周期、变化上报、多 ingress 并存 同时需要目录清晰
#include <softbus_edge/config/EdgeConfig.h>
#include <softbus_edge/egress/CommandExecutor.h>
#include <softbus_edge/egress/ShadowChannelStore.h>
#include <softbus_edge/egress/UplinkClient.h>
#include <softbus_edge/ingress/MockModbusIngress.h>
#include <softbus_edge/runtime/EdgeRuntime.h>
#include <softbus_core/identity/DeviceIdentity.h>
#include <softbus_core/logging/Logger.h>
#include <softbus_core/models/SoftbusTypes.h>
#include <softbus_core/platform/PlatformSignal.h>
#include <softbus_core/time/HwClock.h>
#include <softbus_registry/PointRegistry.h>
#include <softbus_core/config/ConfigWatcher.h>
#include <chrono>
#include <filesystem>
#include <iostream>
#include <string>
#include <thread>
@@ -19,8 +19,10 @@
namespace {
struct EdgeOptions {
std::string configPath{"config/edge.json"};
std::string objectRef{"DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature"};
std::string configPath{"config/edge/edge.json"};
std::string mockObjectRef{
"DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature"};
bool useMock{false};
};
EdgeOptions parseArgs(int argc, char* argv[])
@@ -31,7 +33,9 @@ EdgeOptions parseArgs(int argc, char* argv[])
if (arg == "--config" && i + 1 < argc) {
options.configPath = argv[++i];
} else if (arg == "--object-ref" && i + 1 < argc) {
options.objectRef = argv[++i];
options.mockObjectRef = argv[++i];
} else if (arg == "--mock") {
options.useMock = true;
}
}
return options;
@@ -49,81 +53,44 @@ int main(int argc, char* argv[])
return 1;
}
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;
softbus::edge::EdgeRuntime runtime;
if (!runtime.init(config, options.useMock, options.mockObjectRef)) {
std::cerr << "Failed to initialize edge runtime" << std::endl;
return 1;
}
softbus::edge::ShadowChannelStore shadowStore;
softbus::edge::UplinkClient uplink(config.daemonHost, config.daemonPort);
softbus::edge::CommandExecutor executor(&registry, &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 "
<< config.daemonHost << ":" << config.daemonPort << std::endl;
return 1;
softbus::core::PlatformSignal::install(nullptr);
// 监听物理通道、物理设备、绑定配置文件的变更,如果变更,则重新加载物理通道、物理设备、绑定配置
softbus::core::ConfigWatcher watcher;
watcher.addWatch(config.physicalChannelsPath);
watcher.addWatch(config.physicalDevicesPath);
if (!config.bindingsPatchPath.empty()) {
watcher.addWatch(config.bindingsPatchPath);
}
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();
watcher.poll();
if (softbus::core::PlatformSignal::reloadRequested()) {
runtime.reloadPhysical();
runtime.reloadBindingsPatch();
softbus::core::PlatformSignal::clearReload();
}
runtime.tick();
const auto now = std::chrono::steady_clock::now();
if (now - lastHeartbeat >= std::chrono::milliseconds(config.heartbeatPeriodMs)) {
uplink.sendHeartbeat(config.deviceId);
if (runtime.isUplinkConnected()) {
runtime.uplink()->sendHeartbeat(config.edgeId);
}
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<uint8_t>(entry.registerAddress >> 8),
static_cast<uint8_t>(entry.registerAddress & 0xFF),
static_cast<uint8_t>(raw >> 8),
static_cast<uint8_t>(raw & 0xFF)
};
uplink.sendRawPacket(feedback);
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
ingress.close();
uplink.disconnect();
runtime.shutdown();
return 0;
}

View File

@@ -1,3 +1,4 @@
# role: monitor 用于监控软总线数据模型,并将其转换为物理设备数据
softbus_package(softbus_monitor)
add_library(softbus_monitor STATIC

View File

@@ -3,6 +3,7 @@
#include <softbus_bus/Bus.h>
#include <chrono>
#include <cstdint>
#include <mutex>
#include <string>
#include <unordered_map>
@@ -11,20 +12,29 @@ namespace softbus::monitor {
class HeartbeatMonitor {
public:
// 默认超时 5 秒:超过该时间未收到心跳则判定离线
explicit HeartbeatMonitor(std::chrono::milliseconds timeout = std::chrono::seconds(5));
void attachBus(softbus::bus::Bus* bus);
void tick();
private:
struct DeviceHeartbeatState {
std::chrono::steady_clock::time_point lastSeen;
std::uint64_t lastHeartbeatNs{0};
};
void onHeartbeat(const softbus::bus::BusMessage& message);
void publishOfflineEvent(const std::string& deviceId);
void publishOfflineEvent(const std::string& deviceId,
std::uint64_t offlineAtNs,
std::uint64_t lastHeartbeatNs);
softbus::bus::Bus* bus_{nullptr};
softbus::bus::Bus::SubscriptionId hbSub_{0};
std::chrono::milliseconds timeout_;
std::mutex mutex_;
std::unordered_map<std::string, std::chrono::steady_clock::time_point> lastSeen_;
std::unordered_map<std::string, DeviceHeartbeatState> devices_;
std::unordered_map<std::string, std::uint64_t> offlineSinceNs_;
};
} // namespace softbus::monitor

View File

@@ -7,6 +7,16 @@
namespace softbus::monitor {
namespace {
struct OfflineRecord {
std::string deviceId;
std::uint64_t offlineAtNs{0};
std::uint64_t lastHeartbeatNs{0};
};
} // namespace
HeartbeatMonitor::HeartbeatMonitor(std::chrono::milliseconds timeout)
: timeout_(timeout)
{
@@ -29,27 +39,36 @@ void HeartbeatMonitor::onHeartbeat(const softbus::bus::BusMessage& message)
return;
}
std::lock_guard<std::mutex> lock(mutex_);
lastSeen_[message.deviceId] = std::chrono::steady_clock::now();
devices_[message.deviceId] = DeviceHeartbeatState{
std::chrono::steady_clock::now(),
softbus::core::HwClock::nowNs(),
};
offlineSinceNs_.erase(message.deviceId);
}
void HeartbeatMonitor::tick()
{
const auto now = std::chrono::steady_clock::now();
std::vector<std::string> offline;
const std::uint64_t offlineAtNs = softbus::core::HwClock::nowNs();
std::vector<OfflineRecord> newlyOffline;
{
std::lock_guard<std::mutex> lock(mutex_);
for (const auto& [deviceId, last] : lastSeen_) {
if (now - last > timeout_) {
offline.push_back(deviceId);
for (const auto& [deviceId, state] : devices_) {
if (now - state.lastSeen <= timeout_ || offlineSinceNs_.find(deviceId) != offlineSinceNs_.end()) {
continue;
}
offlineSinceNs_[deviceId] = offlineAtNs;
newlyOffline.push_back(OfflineRecord{deviceId, offlineAtNs, state.lastHeartbeatNs});
}
}
for (const auto& deviceId : offline) {
publishOfflineEvent(deviceId);
for (const auto& record : newlyOffline) {
publishOfflineEvent(record.deviceId, record.offlineAtNs, record.lastHeartbeatNs);
}
}
void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId)
void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId,
std::uint64_t offlineAtNs,
std::uint64_t lastHeartbeatNs)
{
if (!bus_) {
return;
@@ -58,13 +77,20 @@ void HeartbeatMonitor::publishOfflineEvent(const std::string& deviceId)
ev.deviceId = deviceId;
ev.eventType = "edge_offline";
ev.message = "heartbeat timeout";
ev.timestampNs = softbus::core::HwClock::nowNs();
ev.timestampNs = offlineAtNs;
ev.lastHeartbeatNs = lastHeartbeatNs;
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);
SB_LOG_WARN("HeartbeatMonitor",
"edge offline: ",
deviceId,
" offlineAtNs=",
offlineAtNs,
" lastHeartbeatNs=",
lastHeartbeatNs);
}
} // namespace softbus::monitor

View File

@@ -1,3 +1,4 @@
# role: opcua 用于将软总线数据模型转换为 OPC UA 数据模型
softbus_package(softbus_opcua)
add_library(softbus_opcua STATIC

View File

@@ -3,6 +3,7 @@
#include <softbus_bus/Bus.h>
#include <softbus_core/plugin/PluginLoader.h>
#include <softbus_opcua/UaModelBinder.h>
#include <softbus_registry/BindingRegistry.h>
#include <softbus_registry/PointRegistry.h>
namespace softbus::pipeline {
@@ -11,7 +12,8 @@ class PipelineEngine {
public:
PipelineEngine(softbus::core::PluginLoader* plugin,
softbus::opcua::UaModelBinder* binder,
softbus::registry::PointRegistry* registry);
softbus::registry::PointRegistry* registry,
softbus::registry::BindingRegistry* bindings = nullptr);
void attachBus(softbus::bus::Bus* bus);
void process(const softbus::core::RawPacket& packet);
@@ -24,6 +26,7 @@ private:
softbus::core::PluginLoader* plugin_{nullptr};
softbus::opcua::UaModelBinder* binder_{nullptr};
softbus::registry::PointRegistry* registry_{nullptr};
softbus::registry::BindingRegistry* bindings_{nullptr};
softbus::bus::Bus* bus_{nullptr};
softbus::bus::Bus::SubscriptionId rawSub_{0};
};

View File

@@ -7,12 +7,20 @@
namespace softbus::pipeline {
namespace {
constexpr const char* kUnboundTopicPrefix = "edge/unbound/raw/";
} // namespace
PipelineEngine::PipelineEngine(softbus::core::PluginLoader* plugin,
softbus::opcua::UaModelBinder* binder,
softbus::registry::PointRegistry* registry)
softbus::registry::PointRegistry* registry,
softbus::registry::BindingRegistry* bindings)
: plugin_(plugin)
, binder_(binder)
, registry_(registry)
, bindings_(bindings)
{
}
@@ -29,9 +37,14 @@ void PipelineEngine::attachBus(softbus::bus::Bus* bus)
void PipelineEngine::onBusMessage(const softbus::bus::BusMessage& message)
{
if (message.type == softbus::bus::BusPayloadType::RawPacket) {
process(message.rawPacket);
if (message.type != softbus::bus::BusPayloadType::RawPacket) {
return;
}
if (message.topic.rfind("edge/unbound/", 0) == 0
|| message.topic.rfind("edge/raw/unbound/", 0) == 0) {
return;
}
process(message.rawPacket);
}
void PipelineEngine::process(const softbus::core::RawPacket& packet)
@@ -62,16 +75,40 @@ void PipelineEngine::process(const softbus::core::RawPacket& packet)
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);
std::optional<softbus::registry::PointTableEntry> entry;
if (bindings_ && registry_) {
const auto binding = bindings_->findByPhysicalPointId(packet.sourceId);
if (!binding) {
if (bus_) {
softbus::bus::BusMessage busMsg;
busMsg.type = softbus::bus::BusPayloadType::RawPacket;
busMsg.rawPacket = packet;
bus_->publish(std::string(kUnboundTopicPrefix) + packet.sourceId, busMsg);
}
SB_LOG_INFO_TRACE("PipelineEngine", packet.traceId,
"Unbound physical point ", packet.sourceId);
return;
}
entry = registry_->findByLogicalPointId(binding->logicalPointId);
if (entry && binding->scale > 0.0) {
double registerValue = 0.0;
if (parsed.message.value.contains("registerValue")) {
registerValue = parsed.message.value["registerValue"].get<double>();
}
parsed.message.value["value"] = registerValue * binding->scale;
parsed.message.id = entry->objectRef;
}
} else if (registry_) {
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);
"RawPacket converted physicalPointId=", packet.sourceId);
}
void PipelineEngine::publishDataPoint(const softbus::core::DOMMessage& message,

View File

@@ -3,6 +3,7 @@ softbus_package(softbus_registry)
add_library(softbus_registry STATIC
src/PointRegistry.cpp
src/PointTableEntry.cpp
src/BindingRegistry.cpp
)
softbus_export_include(softbus_registry "${CMAKE_CURRENT_SOURCE_DIR}/include")

View File

@@ -0,0 +1,21 @@
#pragma once
#include <string>
namespace softbus::registry {
struct BindingEntry {
std::string bindingId;
std::string edgeId;
std::string physicalPointId;
std::string logicalPointId;
double scale{0.0};
bool enabled{true};
};
struct BindingRoute {
std::string edgeId;
std::string physicalPointId;
};
} // namespace softbus::registry

View File

@@ -0,0 +1,38 @@
#pragma once
#include <softbus_registry/BindingEntry.h>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
namespace softbus::registry {
class BindingRegistry {
public:
bool loadGlobal(const std::string& path);
bool loadLocal(const std::string& path, const std::string& expectedEdgeId = {});
bool reloadGlobal(const std::string& path);
bool applyPatch(const std::string& path);
const std::vector<BindingEntry>& entries() const { return entries_; }
std::optional<BindingEntry> findByBindingId(const std::string& bindingId) const;
std::optional<std::string> resolveLogical(const std::string& physicalPointId) const;
std::optional<BindingRoute> resolveRoute(const std::string& logicalPointId) const;
std::optional<BindingEntry> findByPhysicalPointId(const std::string& physicalPointId) const;
std::optional<std::string> resolvePhysicalLocal(const std::string& logicalPointId) const;
private:
void indexEntry(const BindingEntry& entry);
void clearIndexes();
std::vector<BindingEntry> entries_;
std::unordered_map<std::string, std::size_t> byBindingId_;
std::unordered_map<std::string, std::size_t> byPhysicalPointId_;
std::unordered_map<std::string, std::size_t> byLogicalPointId_;
bool localMode_{false};
};
} // namespace softbus::registry

View File

@@ -3,6 +3,8 @@
#include <softbus_core/models/ObjectModel.h>
#include <softbus_registry/PointTableEntry.h>
#include <nlohmann/json_fwd.hpp>
#include <optional>
#include <string>
#include <unordered_map>
@@ -13,12 +15,17 @@ namespace softbus::registry {
class PointRegistry {
public:
bool loadPointTable(const std::string& path);
bool loadLogicalPoints(const std::string& path);
bool reloadPointTable(const std::string& path);
bool loadTopicRules(const std::string& path);
std::vector<PointTableEntry> diff(const PointRegistry& previous) const;
const softbus::core::ObjectModelTree& tree() const { return tree_; }
const TopicRules& topicRules() const { return topicRules_; }
const std::vector<PointTableEntry>& entries() const { return entries_; }
std::optional<PointTableEntry> findByLogicalPointId(const std::string& logicalPointId) const;
std::optional<PointTableEntry> findByObjectRef(const std::string& objectRef) const;
std::optional<PointTableEntry> findByTopic(const std::string& topic) const;
std::optional<PointTableEntry> findByOpcuaNodeId(const std::string& nodeId) const;
@@ -28,7 +35,10 @@ public:
bool validateWriteValue(const PointTableEntry& entry, double value, std::string* error) const;
private:
bool loadPointTableLegacy(const std::string& path);
bool loadFromJson(const nlohmann::json& j, bool logicalOnly); // NOLINT
void indexEntry(const PointTableEntry& entry);
void rebuildIndexes();
softbus::core::ObjectModelTree tree_;
TopicRules topicRules_;

View File

@@ -0,0 +1,200 @@
#include <softbus_registry/BindingRegistry.h>
#include <fstream>
#include <nlohmann/json.hpp>
namespace softbus::registry {
namespace {
bool parseBindingEntry(const nlohmann::json& j, BindingEntry& entry, bool requireEdgeId)
{
entry.bindingId = j.value("bindingId", "");
entry.edgeId = j.value("edgeId", "");
entry.physicalPointId = j.value("physicalPointId", "");
entry.logicalPointId = j.value("logicalPointId", "");
entry.scale = j.value("scale", 0.0);
entry.enabled = j.value("enabled", true);
if (entry.bindingId.empty() || entry.physicalPointId.empty() || entry.logicalPointId.empty()) {
return false;
}
if (requireEdgeId && entry.edgeId.empty()) {
return false;
}
return true;
}
} // namespace
bool BindingRegistry::loadGlobal(const std::string& path)
{
std::ifstream ifs(path);
if (!ifs) {
return false;
}
nlohmann::json j;
ifs >> j;
localMode_ = false;
entries_.clear();
clearIndexes();
if (!j.contains("bindings")) {
return true;
}
for (const auto& b : j["bindings"]) {
BindingEntry entry;
if (!parseBindingEntry(b, entry, true)) {
continue;
}
entries_.push_back(entry);
indexEntry(entries_.back());
}
return true;
}
bool BindingRegistry::loadLocal(const std::string& path, const std::string& expectedEdgeId)
{
std::ifstream ifs(path);
if (!ifs) {
return false;
}
nlohmann::json j;
ifs >> j;
localMode_ = true;
entries_.clear();
clearIndexes();
const std::string edgeId = j.value("edgeId", expectedEdgeId);
if (!j.contains("bindings")) {
return true;
}
for (const auto& b : j["bindings"]) {
BindingEntry entry;
if (!parseBindingEntry(b, entry, false)) {
continue;
}
entry.edgeId = edgeId;
entries_.push_back(entry);
indexEntry(entries_.back());
}
return true;
}
bool BindingRegistry::reloadGlobal(const std::string& path)
{
return loadGlobal(path);
}
bool BindingRegistry::applyPatch(const std::string& path)
{
std::ifstream ifs(path);
if (!ifs) {
return false;
}
nlohmann::json j;
ifs >> j;
const std::string op = j.value("op", "upsert");
if (!j.contains("binding")) {
return false;
}
BindingEntry entry;
if (!parseBindingEntry(j["binding"], entry, !localMode_)) {
return false;
}
if (localMode_ && entry.edgeId.empty()) {
entry.edgeId = j.value("edgeId", "");
}
if (op == "delete") {
for (auto it = entries_.begin(); it != entries_.end(); ++it) {
if (it->bindingId == entry.bindingId) {
entries_.erase(it);
clearIndexes();
for (const auto& e : entries_) {
indexEntry(e);
}
return true;
}
}
return false;
}
for (auto& e : entries_) {
if (e.bindingId == entry.bindingId) {
e = entry;
clearIndexes();
for (const auto& item : entries_) {
indexEntry(item);
}
return true;
}
}
entries_.push_back(entry);
indexEntry(entries_.back());
return true;
}
void BindingRegistry::clearIndexes()
{
byBindingId_.clear();
byPhysicalPointId_.clear();
byLogicalPointId_.clear();
}
void BindingRegistry::indexEntry(const BindingEntry& entry)
{
const std::size_t idx = entries_.size() - 1;
byBindingId_[entry.bindingId] = idx;
if (entry.enabled) {
byPhysicalPointId_[entry.physicalPointId] = idx;
byLogicalPointId_[entry.logicalPointId] = idx;
}
}
std::optional<BindingEntry> BindingRegistry::findByBindingId(const std::string& bindingId) const
{
const auto it = byBindingId_.find(bindingId);
if (it == byBindingId_.end()) {
return std::nullopt;
}
return entries_[it->second];
}
std::optional<BindingEntry> BindingRegistry::findByPhysicalPointId(
const std::string& physicalPointId) const
{
const auto it = byPhysicalPointId_.find(physicalPointId);
if (it == byPhysicalPointId_.end()) {
return std::nullopt;
}
return entries_[it->second];
}
std::optional<std::string> BindingRegistry::resolveLogical(const std::string& physicalPointId) const
{
const auto entry = findByPhysicalPointId(physicalPointId);
if (!entry) {
return std::nullopt;
}
return entry->logicalPointId;
}
std::optional<BindingRoute> BindingRegistry::resolveRoute(const std::string& logicalPointId) const
{
const auto it = byLogicalPointId_.find(logicalPointId);
if (it == byLogicalPointId_.end()) {
return std::nullopt;
}
const auto& entry = entries_[it->second];
BindingRoute route;
route.edgeId = entry.edgeId;
route.physicalPointId = entry.physicalPointId;
return route;
}
std::optional<std::string> BindingRegistry::resolvePhysicalLocal(const std::string& logicalPointId) const
{
const auto route = resolveRoute(logicalPointId);
if (!route) {
return std::nullopt;
}
return route->physicalPointId;
}
} // namespace softbus::registry

View File

@@ -4,10 +4,16 @@
#include <fstream>
#include <nlohmann/json.hpp>
#include <unordered_set>
namespace softbus::registry {
bool PointRegistry::loadPointTable(const std::string& path)
{
return loadPointTableLegacy(path);
}
bool PointRegistry::loadPointTableLegacy(const std::string& path)
{
std::ifstream ifs(path);
if (!ifs) {
@@ -15,16 +21,56 @@ bool PointRegistry::loadPointTable(const std::string& path)
}
nlohmann::json j;
ifs >> j;
if (!loadFromJson(j, false)) {
return false;
}
rebuildIndexes();
return true;
}
bool PointRegistry::loadLogicalPoints(const std::string& path)
{
std::ifstream ifs(path);
if (!ifs) {
return false;
}
nlohmann::json j;
ifs >> j;
if (!loadFromJson(j, true)) {
return false;
}
rebuildIndexes();
return true;
}
bool PointRegistry::reloadPointTable(const std::string& path)
{
std::ifstream ifs(path);
if (!ifs) {
return false;
}
nlohmann::json j;
ifs >> j;
bool logicalOnly = true;
if (j.contains("devices") && j["devices"].is_array() && !j["devices"].empty()
&& j["devices"][0].contains("points") && j["devices"][0]["points"].is_array()
&& !j["devices"][0]["points"].empty()) {
logicalOnly = !j["devices"][0]["points"][0].contains("register");
}
if (!loadFromJson(j, logicalOnly)) {
return false;
}
rebuildIndexes();
return true;
}
bool PointRegistry::loadFromJson(const nlohmann::json& j, bool logicalOnly)
{
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();
@@ -32,9 +78,12 @@ bool PointRegistry::loadPointTable(const std::string& path)
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);
device.stableDeviceKey = dev.value(
"stableDeviceKey",
dev.value("logicalDeviceId", dev.value("deviceId", "")));
device.protocol = dev.value("protocol", logicalOnly ? "logical" : "modbus");
const std::string deviceId = dev.value(
"deviceId", dev.value("logicalDeviceId", device.stableDeviceKey));
if (!dev.contains("points")) {
tree_.devices.push_back(std::move(device));
@@ -43,13 +92,13 @@ bool PointRegistry::loadPointTable(const std::string& path)
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.name = pt.value("name", pt.value("logicalPointId", pt.value("pointId", "")));
pointDef.registerAddress = logicalOnly ? 0 : pt.value("register", 0);
pointDef.domain = pt.value("domain", "Process");
pointDef.signalType = pt.value("signalType", "AI");
pointDef.physicalChannel = pt.value("physicalChannel", "");
pointDef.physicalChannel = logicalOnly ? "" : pt.value("physicalChannel", "");
pointDef.unit = pt.value("unit", "");
pointDef.scale = pt.value("scale", 1.0);
pointDef.scale = logicalOnly ? 1.0 : 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) {
@@ -63,7 +112,7 @@ bool PointRegistry::loadPointTable(const std::string& path)
PointTableEntry entry;
entry.deviceId = deviceId;
entry.pointId = pt.value("pointId", pointDef.name);
entry.pointId = pt.value("pointId", pt.value("logicalPointId", pointDef.name));
entry.name = pointDef.name;
entry.displayName = pt.value("displayName", entry.name);
entry.signalType = pointDef.signalType;
@@ -80,8 +129,7 @@ bool PointRegistry::loadPointTable(const std::string& path)
tree_, device.stableDeviceKey, entry.name);
entry.objectRef = objectRef.toPath();
entry.opcuaNodeId = pt.value(
"opcuaNodeId", "ns=1;s=" + entry.objectRef);
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(
@@ -90,13 +138,27 @@ bool PointRegistry::loadPointTable(const std::string& path)
"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();
}
std::vector<PointTableEntry> PointRegistry::diff(const PointRegistry& previous) const
{
std::unordered_set<std::string> previousKeys;
for (const auto& e : previous.entries_) {
previousKeys.insert(e.deviceId + "/" + e.pointId);
}
std::vector<PointTableEntry> added;
for (const auto& e : entries_) {
if (previousKeys.find(e.deviceId + "/" + e.pointId) == previousKeys.end()) {
added.push_back(e);
}
}
return added;
}
bool PointRegistry::loadTopicRules(const std::string& path)
{
std::ifstream ifs(path);
@@ -113,6 +175,29 @@ bool PointRegistry::loadTopicRules(const std::string& path)
return true;
}
void PointRegistry::rebuildIndexes()
{
byObjectRef_.clear();
byTopic_.clear();
byOpcuaNodeId_.clear();
byDevicePoint_.clear();
for (std::size_t i = 0; i < entries_.size(); ++i) {
const auto& entry = entries_[i];
byObjectRef_[entry.objectRef] = i;
byOpcuaNodeId_[entry.opcuaNodeId] = i;
byDevicePoint_[entry.deviceId + "/" + entry.pointId] = i;
if (!entry.dataTopic.empty()) {
byTopic_[entry.dataTopic] = i;
}
if (!entry.cmdTopic.empty()) {
byTopic_[entry.cmdTopic] = i;
}
if (!entry.ackTopic.empty()) {
byTopic_[entry.ackTopic] = i;
}
}
}
void PointRegistry::indexEntry(const PointTableEntry& entry)
{
const std::size_t idx = entries_.size() - 1;
@@ -130,6 +215,18 @@ void PointRegistry::indexEntry(const PointTableEntry& entry)
}
}
std::optional<PointTableEntry> PointRegistry::findByLogicalPointId(
const std::string& logicalPointId) const
{
const auto slash = logicalPointId.find('/');
if (slash == std::string::npos) {
return std::nullopt;
}
return findByDevicePoint(
logicalPointId.substr(0, slash),
logicalPointId.substr(slash + 1));
}
std::optional<PointTableEntry> PointRegistry::findByObjectRef(const std::string& objectRef) const
{
const auto it = byObjectRef_.find(objectRef);

View File

@@ -26,6 +26,7 @@ public:
bool sendRawPacket(const softbus::core::RawPacket& packet);
bool sendAck(const softbus::core::SoftbusAck& ack);
bool sendHeartbeat(const std::string& deviceId);
bool sendEdgeRegister(const std::string& edgeId);
void poll();
bool isConnected() const { return connected_; }

View File

@@ -8,8 +8,7 @@
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include <unordered_map>
namespace softbus::transport {
@@ -27,18 +26,28 @@ public:
void start();
void stop();
bool sendCommand(const softbus::api::ExecuteCommand& command);
bool sendCommandToEdge(const std::string& edgeId, const softbus::api::ExecuteCommand& command);
bool hasEdge(const std::string& edgeId) const;
private:
struct EdgeSession {
std::shared_ptr<asio::ip::tcp::socket> socket;
std::string edgeId;
std::string readBuffer;
};
void doAccept();
void doRead(std::shared_ptr<asio::ip::tcp::socket> socket);
void handleLine(const std::string& line);
void doRead(const std::shared_ptr<EdgeSession>& session);
void handleLine(const std::shared_ptr<EdgeSession>& session, const std::string& line);
void removeSession(const std::shared_ptr<EdgeSession>& session);
asio::io_context& io_;
asio::ip::tcp::acceptor acceptor_;
RawPacketHandler rawHandler_;
WireMessageHandler wireHandler_;
std::shared_ptr<asio::ip::tcp::socket> clientSocket_;
std::string readBuffer_;
mutable std::mutex sessionMutex_;
std::vector<std::shared_ptr<EdgeSession>> sessions_;
std::unordered_map<std::string, std::shared_ptr<EdgeSession>> edgeSessions_;
std::mutex sendMutex_;
bool running_{false};
};

View File

@@ -17,6 +17,7 @@ enum class WireMessageType {
ExecuteCommand,
Ack,
Heartbeat,
EdgeRegister,
Unknown
};
@@ -26,12 +27,14 @@ struct WireMessage {
softbus::api::ExecuteCommand command;
softbus::core::SoftbusAck ack;
std::string heartbeatDeviceId;
std::string edgeId;
};
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::string encodeEdgeRegister(const std::string& edgeId);
std::optional<WireMessage> decodeWireMessage(const std::string& line);
std::string base64Encode(const uint8_t* data, std::size_t size);

View File

@@ -86,6 +86,18 @@ bool TcpUplinkClient::sendAck(const softbus::core::SoftbusAck& ack)
return !ec;
}
bool TcpUplinkClient::sendEdgeRegister(const std::string& edgeId)
{
if (!socket_ || !socket_->is_open()) {
return false;
}
const auto line = encodeEdgeRegister(edgeId);
std::lock_guard<std::mutex> 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()) {

View File

@@ -2,6 +2,7 @@
#include <softbus_core/logging/Logger.h>
#include <algorithm>
#include <array>
namespace softbus::transport {
@@ -41,77 +42,133 @@ void TcpUplinkServer::stop()
running_ = false;
std::error_code ec;
acceptor_.close(ec);
if (clientSocket_) {
clientSocket_->close(ec);
clientSocket_.reset();
std::lock_guard<std::mutex> lock(sessionMutex_);
for (auto& session : sessions_) {
if (session->socket) {
session->socket->close(ec);
}
}
sessions_.clear();
edgeSessions_.clear();
}
bool TcpUplinkServer::sendCommand(const softbus::api::ExecuteCommand& command)
{
if (!clientSocket_ || !clientSocket_->is_open()) {
std::string edgeId;
{
std::lock_guard<std::mutex> lock(sessionMutex_);
if (!edgeSessions_.empty()) {
edgeId = edgeSessions_.begin()->first;
}
}
if (!edgeId.empty()) {
return sendCommandToEdge(edgeId, command);
}
std::shared_ptr<EdgeSession> session;
{
std::lock_guard<std::mutex> lock(sessionMutex_);
if (sessions_.empty() || !sessions_.front()->socket) {
return false;
}
session = sessions_.front();
}
const auto line = encodeWireMessage(command);
std::lock_guard<std::mutex> sendLock(sendMutex_);
std::error_code ec;
asio::write(*session->socket, asio::buffer(line), ec);
return !ec;
}
bool TcpUplinkServer::sendCommandToEdge(const std::string& edgeId,
const softbus::api::ExecuteCommand& command)
{
std::shared_ptr<EdgeSession> session;
{
std::lock_guard<std::mutex> lock(sessionMutex_);
const auto it = edgeSessions_.find(edgeId);
if (it == edgeSessions_.end()) {
return false;
}
session = it->second;
}
if (!session || !session->socket || !session->socket->is_open()) {
return false;
}
const auto line = encodeWireMessage(command);
std::lock_guard<std::mutex> lock(sendMutex_);
std::lock_guard<std::mutex> sendLock(sendMutex_);
std::error_code ec;
asio::write(*clientSocket_, asio::buffer(line), ec);
asio::write(*session->socket, asio::buffer(line), ec);
return !ec;
}
bool TcpUplinkServer::hasEdge(const std::string& edgeId) const
{
std::lock_guard<std::mutex> lock(sessionMutex_);
return edgeSessions_.find(edgeId) != edgeSessions_.end();
}
void TcpUplinkServer::doAccept()
{
if (!running_) {
return;
}
auto socket = std::make_shared<asio::ip::tcp::socket>(io_);
acceptor_.async_accept(*socket, [this, socket](const std::error_code& ec) {
auto session = std::make_shared<EdgeSession>();
session->socket = std::make_shared<asio::ip::tcp::socket>(io_);
acceptor_.async_accept(*session->socket, [this, session](const std::error_code& ec) {
if (ec) {
if (running_) {
doAccept();
}
return;
}
clientSocket_ = socket;
readBuffer_.clear();
SB_LOG_INFO("TcpUplinkServer", "Edge connected");
doRead(socket);
{
std::lock_guard<std::mutex> lock(sessionMutex_);
sessions_.push_back(session);
}
SB_LOG_INFO("TcpUplinkServer", "Edge connected (awaiting edgeRegister)");
doRead(session);
doAccept();
});
}
void TcpUplinkServer::doRead(const std::shared_ptr<asio::ip::tcp::socket> socket)
void TcpUplinkServer::doRead(const std::shared_ptr<EdgeSession>& session)
{
auto buffer = std::make_shared<std::array<char, 4096>>();
socket->async_read_some(asio::buffer(*buffer),
[this, socket, buffer](const std::error_code& ec, std::size_t bytes) {
session->socket->async_read_some(asio::buffer(*buffer),
[this, session, buffer](const std::error_code& ec, std::size_t bytes) {
if (ec) {
SB_LOG_WARN("TcpUplinkServer", "Edge disconnected");
if (clientSocket_ == socket) {
clientSocket_.reset();
}
SB_LOG_WARN("TcpUplinkServer", "Edge disconnected edgeId=", session->edgeId);
removeSession(session);
return;
}
readBuffer_.append(buffer->data(), bytes);
session->readBuffer.append(buffer->data(), bytes);
for (;;) {
const auto pos = readBuffer_.find('\n');
const auto pos = session->readBuffer.find('\n');
if (pos == std::string::npos) {
break;
}
const std::string line = readBuffer_.substr(0, pos);
readBuffer_.erase(0, pos + 1);
handleLine(line);
const std::string line = session->readBuffer.substr(0, pos);
session->readBuffer.erase(0, pos + 1);
handleLine(session, line);
}
doRead(socket);
doRead(session);
});
}
void TcpUplinkServer::handleLine(const std::string& line)
void TcpUplinkServer::handleLine(const std::shared_ptr<EdgeSession>& session,
const std::string& line)
{
const auto msg = decodeWireMessage(line);
if (!msg) {
return;
}
if (msg->type == WireMessageType::EdgeRegister && !msg->edgeId.empty()) {
session->edgeId = msg->edgeId;
std::lock_guard<std::mutex> lock(sessionMutex_);
edgeSessions_[msg->edgeId] = session;
SB_LOG_INFO("TcpUplinkServer", "Edge registered edgeId=", msg->edgeId);
return;
}
if (wireHandler_) {
wireHandler_(*msg);
}
@@ -120,4 +177,15 @@ void TcpUplinkServer::handleLine(const std::string& line)
}
}
void TcpUplinkServer::removeSession(const std::shared_ptr<EdgeSession>& session)
{
std::lock_guard<std::mutex> lock(sessionMutex_);
if (!session->edgeId.empty()) {
edgeSessions_.erase(session->edgeId);
}
sessions_.erase(
std::remove(sessions_.begin(), sessions_.end(), session),
sessions_.end());
}
} // namespace softbus::transport

View File

@@ -29,7 +29,14 @@ void TransportBusBridge::onBusMessage(const softbus::bus::BusMessage& message)
if (!server_ || message.type != softbus::bus::BusPayloadType::ExecuteCommand) {
return;
}
if (!server_->sendCommand(message.command)) {
const std::string edgeId = message.command.params.value("edgeId", "");
bool sent = false;
if (!edgeId.empty()) {
sent = server_->sendCommandToEdge(edgeId, message.command);
} else {
sent = server_->sendCommand(message.command);
}
if (!sent) {
SB_LOG_WARN("TransportBusBridge", "failed to send command to edge topic=", message.topic);
}
}

View File

@@ -132,7 +132,21 @@ std::optional<WireMessage> decodeWireMessage(const std::string& line)
msg.heartbeatDeviceId = j.value("deviceId", "");
return msg;
}
if (type == "edgeRegister") {
msg.type = WireMessageType::EdgeRegister;
msg.edgeId = j.value("edgeId", "");
return msg;
}
return std::nullopt;
}
std::string encodeEdgeRegister(const std::string& edgeId)
{
nlohmann::json j{
{"type", "edgeRegister"},
{"edgeId", edgeId}
};
return j.dump() + "\n";
}
} // namespace softbus::transport

View File

@@ -7,12 +7,13 @@ 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 \
--config config/daemon/daemon_profile.json \
--logical-points config/daemon/logical_points.json \
--bindings-global config/daemon/bindings_global.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 &
./build/bin/softbus_edge --config config/edge/edge.json > /tmp/valve_loop_edge.log 2>&1 &
EDGE_PID=$!
sleep 8
@@ -21,6 +22,6 @@ 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
grep -qE "write edge01/phy_modbus_s1_r40002|ValveOpening value=60" /tmp/valve_loop_edge.log
echo "valve_control_loop: PASS"

13
third_party/README.md vendored
View File

@@ -5,6 +5,7 @@
| 目录 | 版本 | 许可证 | 用途 |
|------|------|--------|------|
| `open62541/` | v1.4.6commit `50ae40d3` | MPL-2.0 | OPC UA 协议栈 |
| `libmodbus/` | v3.1.10 | LGPL-2.1+ | edge Modbus RTU/TCP |
## open62541
@@ -17,3 +18,15 @@
1. 用新 tag 替换 `third_party/open62541/` 目录内容(保留 `LICENSE` 等上游文件)
2. 更新本表版本号与 commit
3. 全量编译并跑 OPC UA 联调(守护进程 + 边缘 + 客户端读写)
## libmodbus
- 上游:<https://github.com/stephane/libmodbus>
-`cmake/libmodbus/CMakeLists.txt` 编译为静态库 `modbus_vendor`,供 `softbus_edge` 链接
- 若系统已安装 `libmodbus-dev`pkg-config 可见),优先使用系统库;可用 `-DSOFTBUS_FORCE_VENDOR_LIBMODBUS=ON` 强制 vendor
### 升级步骤
1. 替换 `third_party/libmodbus/` 为新 tag
2. 更新本表版本号
3. 重新编译 `softbus_edge` 并验证串口读写

1
third_party/libmodbus vendored Submodule

Submodule third_party/libmodbus added at 2cbafa3113

View File

@@ -9,3 +9,6 @@ target_link_libraries(topic_watcher PRIVATE
softbus_bus softbus_core
)
softbus_apply_platform_libs(topic_watcher)
add_executable(modbus_scanner modbus_scanner.cpp)
softbus_apply_platform_libs(modbus_scanner)

27
tools/modbus_scanner.cpp Normal file
View File

@@ -0,0 +1,27 @@
#include <iostream>
#include <string>
int main(int argc, char* argv[])
{
std::string port = "/dev/ttyUSB0";
int baud = 9600;
int slaveStart = 1;
int slaveEnd = 10;
for (int i = 1; i < argc; ++i) {
const std::string arg = argv[i];
if (arg == "--port" && i + 1 < argc) {
port = argv[++i];
} else if (arg == "--baud" && i + 1 < argc) {
baud = std::stoi(argv[++i]);
} else if (arg == "--slave-start" && i + 1 < argc) {
slaveStart = std::stoi(argv[++i]);
} else if (arg == "--slave-end" && i + 1 < argc) {
slaveEnd = std::stoi(argv[++i]);
}
}
std::cout << "modbus_scanner: draft output for physical_devices.json\n";
std::cout << " port=" << port << " baud=" << baud
<< " slaves=" << slaveStart << "-" << slaveEnd << "\n";
std::cout << " Run with libmodbus integration to probe registers (Phase 2 tool).\n";
return 0;
}