diff --git a/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx b/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx index 51b6cfb..eebf2be 100644 Binary files a/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx and b/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx differ diff --git a/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx b/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx index f579254..b7c46dd 100644 Binary files a/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx and b/.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx differ diff --git a/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx b/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx index 573bfe2..82ae9d9 100644 Binary files a/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx and b/.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx differ diff --git a/.cache/clangd/index/DeviceTreeModel.cpp.D68EA11C8BBF4ADF.idx b/.cache/clangd/index/DeviceTreeModel.cpp.D68EA11C8BBF4ADF.idx index 3718a04..a6b9b28 100644 Binary files a/.cache/clangd/index/DeviceTreeModel.cpp.D68EA11C8BBF4ADF.idx and b/.cache/clangd/index/DeviceTreeModel.cpp.D68EA11C8BBF4ADF.idx differ diff --git a/.cache/clangd/index/DeviceTreeModel.h.A5637DBA59AB7F8A.idx b/.cache/clangd/index/DeviceTreeModel.h.A5637DBA59AB7F8A.idx index 8bf1425..36a3112 100644 Binary files a/.cache/clangd/index/DeviceTreeModel.h.A5637DBA59AB7F8A.idx and b/.cache/clangd/index/DeviceTreeModel.h.A5637DBA59AB7F8A.idx differ diff --git a/.cache/clangd/index/IpcDBus.cpp.F75762A8BF6179E0.idx b/.cache/clangd/index/IpcDBus.cpp.F75762A8BF6179E0.idx index 83a7e4f..067eb85 100644 Binary files a/.cache/clangd/index/IpcDBus.cpp.F75762A8BF6179E0.idx and b/.cache/clangd/index/IpcDBus.cpp.F75762A8BF6179E0.idx differ diff --git a/CMakeLists.txt b/CMakeLists.txt index 6093419..7e8c299 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -31,6 +31,13 @@ set(CMAKE_AUTORCC ON) set(CMAKE_AUTOMOC ON) # Qt6：daemon 使用 Core/DBus/Network/SerialPort/Sql（orchestrator 已迁至 Python） find_package(Qt6 6.5 REQUIRED COMPONENTS Core SerialPort Network Sql DBus) +find_package(Qt6 QUIET COMPONENTS SerialBus) +if(TARGET Qt6::SerialBus) + set(SOFTBUS_DAEMON_HAVE_CAN 1) +else() + set(SOFTBUS_DAEMON_HAVE_CAN 0) + message(STATUS "Qt6 SerialBus not found: building softbus_daemon without local SocketCAN devices") +endif() include(CheckIncludeFileCXX) @@ -182,6 +189,10 @@ qt_add_executable(softbus_daemon src/message_bus/ingress/IIngressPort.h src/message_bus/ingress/DriverIngressAdapter.h src/message_bus/ingress/DriverIngressAdapter.cpp + src/message_bus/ingress/EdgeUplinkWire.h + src/message_bus/ingress/EdgeUplinkWire.cpp + src/message_bus/ingress/EdgeTcpIngressService.h + src/message_bus/ingress/EdgeTcpIngressService.cpp src/message_bus/egress/IEgressPort.h src/message_bus/egress/EgressRouter.h src/message_bus/egress/EgressRouter.cpp @@ -201,6 +212,29 @@ target_link_libraries(softbus_daemon Qt6::Network ) +if(SOFTBUS_DAEMON_HAVE_CAN) + target_sources(softbus_daemon PRIVATE + src/device_bus/manager/CanDeviceManager.cpp + src/device_bus/manager/CanDeviceManager.h + src/devices/physical/CanDevice.cpp + src/devices/physical/CanDevice.h + ) + target_link_libraries(softbus_daemon PRIVATE Qt6::SerialBus) +endif() + +target_compile_definitions(softbus_daemon PRIVATE SOFTBUS_DAEMON_HAVE_CAN=${SOFTBUS_DAEMON_HAVE_CAN}) + +qt_add_executable(edge_uplink_wire_test + tests/edge_uplink_wire_test.cpp + src/message_bus/ingress/EdgeUplinkWire.cpp +) +target_include_directories(edge_uplink_wire_test + PRIVATE + ${CMAKE_SOURCE_DIR} + ${CMAKE_SOURCE_DIR}/src +) +target_link_libraries(edge_uplink_wire_test PRIVATE Qt6::Core) + # Export daemon symbols so runtime-loaded protocol plugins can resolve # shared singleton APIs (e.g. DiscoveryPool/MappingRegistry) from host process. # if(UNIX AND NOT APPLE) diff --git a/bin/edge_uplink_wire_test b/bin/edge_uplink_wire_test new file mode 100644 index 0000000..e750db1 Binary files /dev/null and b/bin/edge_uplink_wire_test differ diff --git a/bin/softbus_daemon b/bin/softbus_daemon index ebf4377..9a74d5f 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/docs/EDGE_LOCAL_SETUP.md b/docs/EDGE_LOCAL_SETUP.md new file mode 100644 index 0000000..df1265a --- /dev/null +++ b/docs/EDGE_LOCAL_SETUP.md @@ -0,0 +1,19 @@ +# 边缘上行与核心本地联调 + +## 前提 + +- 已编译 `softbus_daemon`（含 `EdgeTcpIngressService`）。 +- 在 `~/softbus/config/daemon_config.json` 中启用 **`edgeIngress.enabled`**，并设置 **`port`**（默认 `18765`）。若设置 **`authToken`**，边缘首帧 HELLO 必须携带相同 token。 +- 设备树中需存在可被边缘修改的节点：满足 `params.edgeId` 等于边缘 `edgeId`，或节点 `id` 为 `edges/` / `edges//...` 前缀（见 `DeviceTreeModel::applyEdgePropose`）。 + +## 快速验证 + +1. 启动核心：`softbus_daemon`（`CoreRouter` 角色，默认）。 +2. 运行单元测试（线格式）：`bin/edge_uplink_wire_test`。 +3. 使用 `edge_agent`（独立仓库）连接核心、发送 HELLO 与可选 `DATA_RAWBUS` / `TREE_PROPOSE`（见该仓库 README）。 + +## D-Bus 辅助命令 + +- `list_edge_sessions`：查看当前边缘 TCP 会话摘要。 +- `get_edge_tree_revision`：当前内存设备树 `treeRevision`。 +- `push_edge_tree`：向指定 `edgeId` 的已连接边缘 TCP 会话发送 **TREE_FULL**（子树 JSON，规则同 `DeviceTreeModel::exportSubtreeForEdge`）。HELLO 成功后核心也会自动推送一次。**`patch_device_tree_node`** 若修改的节点属于某边缘作用域（`params.edgeId` 或 `edges//...`），响应里会带 **`edgeTreeDownlink`** 并尝试对该 edge 自动下行（无会话时 `ok=false` 等见各元素 `error` 字段）。 diff --git a/docs/EDGE_UPLINK_PROTOCOL_v1.md b/docs/EDGE_UPLINK_PROTOCOL_v1.md new file mode 100644 index 0000000..7ec154e --- /dev/null +++ b/docs/EDGE_UPLINK_PROTOCOL_v1.md @@ -0,0 +1,146 @@ +# SoftBus 边缘上行协议 v1（EDGE_UPLINK_PROTOCOL_v1） + +本文档定义 **边缘 `edge_agent` → 核心 `softbus_daemon`** 的 TCP 二进制帧格式，支持： + +- **数据面**：`DATA_RAWBUS`（载荷语义对齐 [`RawBusMessage`](file:///home/dt/myCode/bus/soft_bus/softbus_daemon/src/core/models/RawBusMessage.h)） +- **控制面**：会话、`CTRL_TREE_*` 与 **反向同步** `CTRL_TREE_PROPOSE` / `CTRL_TREE_RESULT` + +边缘仓库应维护与本文档 **字节级一致** 的副本（如 `docs/UPLINK_PROTOCOL_v1.md`）。 + +## 1. 传输 + +- **传输层**：TCP，由核心监听（`edgeIngress.listenAddress` / `port`），边缘主动连接。 +- **字节序**：多字节整数均为 **大端（网络序）**。 +- **帧大小上限**：由核心配置 `maxFrameBytes`（默认 4MiB）强制；超限则断开连接。 + +## 2. 外层帧（所有 `frame_kind` 共用） + +| 偏移 | 长度 | 字段 | 说明 | +|------|------|------|------| +| 0 | 4 | magic | 固定 ASCII `SBUP`（0x53 0x42 0x55 0x50） | +| 4 | 2 | wire_version | 当前为 **1** | +| 6 | 2 | frame_kind | 见 §3 | +| 8 | 4 | reserved | 填 0 | +| 12 | 4 | payload_len | 后续 payload 字节数 | +| 16 | payload_len | payload | 依 `frame_kind` 解析 | + +## 3. frame_kind 枚举（u16） + +| 值 | 名称 | 方向 | 说明 | +|---:|------|------|------| +| 1 | HELLO | 边缘→核心 | UTF-8 JSON，见 §4 | +| 2 | ACK | 边缘→核心 | UTF-8 JSON，可选（应用下行树后确认） | +| 3 | DATA_RAWBUS | 边缘→核心 | 二进制，见 §5 | +| 4 | TREE_FULL | 核心→边缘 | UTF-8 JSON，见 §4.1 | +| 5 | TREE_PATCH | 核心→边缘 | UTF-8 JSON（预留，下行） | +| 6 | TREE_PROPOSE | 边缘→核心 | UTF-8 JSON，见 §6 | +| 7 | TREE_RESULT | 核心→边缘 | UTF-8 JSON，见 §7 | +| 8 | REPORT | 边缘→核心 | UTF-8 JSON（预留） | + +## 4. HELLO（边缘→核心） + +若核心配置了 `authToken`（非空），则 **首帧必须是 HELLO**，否则核心断开。 + +JSON 字段： + +- `edgeId`（string，必填）：与会话绑定；后续 `TREE_PROPOSE` 须与此一致。 +- `authToken`（string，若核心配置了 token 则必填且相等）。 +- `currentTreeRevision`（integer，可选）：边缘当前已应用的 `treeRevision`，供核心决定是否下行全量。 + +核心响应：可紧跟发送 `TREE_RESULT`；**HELLO 成功后**核心会再发送一帧 **`TREE_FULL`**（与 `push_edge_tree` 下行格式相同，见 §4.1）。 + +## 4.1 TREE_FULL（核心→边缘） + +UTF-8 JSON，置于外层 `frame_kind=4` 的 payload 中： + +- `kind`：固定字符串 `"TREE_FULL"`（便于与纯数组载荷区分）。 +- `edgeId`（string）：与会话绑定，边缘应校验与本地 `edgeId` 一致。 +- `treeRevision`（integer）：导出子树时刻的**全局** `treeRevision`（与 `TREE_PROPOSE` 的 `baseTreeRevision` 语义一致）。 +- `tree`（array）：节点对象数组，字段形状与核心 `get_device_tree` 中单节点一致（`id`/`name`/`type`/`endpoint`/`driver`/`capabilitiesRef`/`params`/`enabled`/`children`）；仅包含 **属于该 edge 作用域** 的节点（规则同 §6.2）。 + +空 `tree` 表示核心当前未为该 edge 分配任何节点（仍合法）。 + +## 5. DATA_RAWBUS 的 payload（二进制） + +在 **外层 payload** 内为定长头 + 变长 `traceId` + 变长 **bus payload**。 + +| 偏移 | 长度 | 字段 | 类型 | +|------|------|------|------| +| 0 | 2 | trace_len | u16 BE，UTF-8 字节数，最大 512 | +| 2 | 4 | device_id | i32 BE | +| 6 | 4 | endpoint_hash | u32 BE | +| 10 | 2 | protocol | u16 BE，与 `ProtocolType` 数值一致 | +| 12 | 2 | reserved0 | u16 BE，0 | +| 14 | 4 | logical_address | u32 BE | +| 18 | 8 | routing_key | u64 BE | +| 26 | 8 | protocol_header | 8 字节原始拷贝至 `ProtocolHeader` union | +| 34 | 8 | timestamp_ms | i64 BE | +| 42 | 1 | direction | i8，`BusDirection` 枚举数值 | +| 43 | 5 | reserved1 | 填 0 | +| 48 | trace_len | trace_utf8 | UTF-8 | +| 48+trace_len | 剩余 | bus_payload | 写入 `PayloadRef` | + +总长度：`48 + trace_len + bus_payload_len` 须等于外层 `payload_len`。 + +### 5.1 SocketCAN / `CAN_RAW`（边缘与核心本机可选） + +- **endpoint 字符串**：建议使用 **`can:`**，例如 `can:vcan0`。设备树节点 `type` 为 `can` 或仅依赖 `endpoint` 前缀推断时，边缘/核心 `EdgeDeviceBus` / `DeviceBusManager` 使用 Qt **SocketCAN** 插件（`socketcan`）打开对应接口；`makeEndpointHash(endpoint)` 与串口一致为 `qHash(endpoint)`。 +- **bus_payload**：单帧 **数据场** 原始字节（0–8 字节）。**DLC=0 的数据帧合法**；核心 `EdgeTcpIngressService` 对 `protocol == CAN_RAW` 或 `CAN_OPEN` 时允许空 `bus_payload`，仍携带 `protocol_header` 中的 CAN id 元数据。 +- **protocol（u16）**：与 `ProtocolType` 枚举数值一致；CAN 上行应使用 **`CAN_RAW`**（设备树 `params.protocol` 常用字符串 `can_raw`，与 `protocolTypeFromHint` 一致）。 +- **protocol_header（8 字节）**：与 `ProtocolHeader::can` 一致：`cobId`（29 位有效）、`isExtended`、`isRtr`；编码侧将 `RawBusMessage::header` 按 8 字节写入再参与 `DATA_RAWBUS` 定长头（与现有 Modbus 头填法相同，见实现 `encodeDataRawBusPayload`）。 + +## 6. TREE_PROPOSE（边缘→核心，反向同步） + +UTF-8 JSON。 + +### 6.1 固定字段 + +- `edgeId`（string）：须与 HELLO 中一致。 +- `baseTreeRevision`（integer）：**全局树 revision**（与核心 `DeviceTreeModel::treeRevision()` 一致）；不一致则 **CONFLICT**。 +- `entries`（array，必填）：每项为下列之一。 + +#### 6.1.1 `patch`（默认，与历史行为一致） + +- `op` 省略或 `"patch"`。 +- `id`（string，必填）：须已存在于设备树。 +- `patch`（object，必填）：字段规则与 D-Bus `patch_device_tree_node` 的 `patch` 相同（`name`/`type`/`endpoint`/`driver`/`capabilitiesRef`/`enabled`/`params` 对象合并）。 + +#### 6.1.2 `upsert`（边缘注册新节点或整节点合并） + +- `op`：`"upsert"`。 +- `id`（string，必填）：目标节点 id。新建时 **必须** 为 `edges/` 或 `edges//...` 前缀（与 §6.2 作用域一致）；已存在节点则按作用域校验后 **合并** `node` 字段。 +- `node`（object，必填）：与 `get_device_tree` 单节点形状一致：`id`/`name`/`type`/`endpoint`/`driver`/`capabilitiesRef`/`params`/`enabled`/`children`（`children` 为子节点 id 字符串数组）。新建时若省略 `node.id` 则使用条目 `id`；若 `node.id` 与条目 `id` 均非空则须相等。 +- **新建**：核心在 `params` 中未带 `edgeId` 时自动写入会话 `edgeId`；合并后节点须满足 §6.2。 +- **`children`**：仅允许引用同一边缘子树内的 id（`edges/` 或其前缀下的子 id），否则 `FORBIDDEN_SCOPE`。 + +### 6.2 作用域（MVP） + +仅允许修改满足以下 **任一** 条件的节点： + +- `node.params.edgeId`（string）等于会话 `edgeId`；或 +- `node.id` 等于 `edges/` 或以 `edges//` 为前缀。 + +否则整批 **拒绝**（`FORBIDDEN_SCOPE`）。 + +### 6.3 并发 + +与 D-Bus `patch_device_tree_node` 共用 **`DeviceTreeModel` 写锁**，保证 revision 单调、无丢失更新。成功应用后，核心还会向该会话 **`edgeId` 的所有已 HELLO 连接** 再发一帧 **`TREE_FULL`**（与 `push_edge_tree` 载荷一致），便于边缘落盘与 revision 对齐。 + +## 7. TREE_RESULT（核心→边缘） + +UTF-8 JSON： + +- `ok`（bool） +- `code`（string）：`OK` | `CONFLICT` | `FORBIDDEN_SCOPE` | `INVALID_JSON` | `NODE_NOT_FOUND` | `VALIDATION_ERROR` +- `message`（string，可选） +- `newTreeRevision`（integer，成功时）：应用后的 revision +- `currentTreeRevision`（integer，CONFLICT 时建议带）：核心当前 revision + +## 8. treeRevision 语义（MVP） + +- 核心在进程内维护单调递增的 **`treeRevision`**（从 1 起，每次成功变更设备树内存模型后 +1）。 +- 守护进程重启后 revision **重置**（MVP）；边缘应在 HELLO 后若收到 `CONFLICT` 或 revision 不明，依赖核心自动下发的 **`TREE_FULL`**、`push_edge_tree` 运维命令，或重新下发配置。 + +## 9. 与 orchestrator / D-Bus + +核心侧设备树持久化与 `reconcile` 行为与现有 `patch_device_tree_node` 一致；本协议为 **额外入口**，不得绕过校验逻辑。 diff --git a/main.cpp b/main.cpp index 82f74d5..cf19a31 100644 --- a/main.cpp +++ b/main.cpp @@ -38,7 +38,9 @@ int main(int argc, char *argv[]) parser.addHelpOption(); parser.addVersionOption(); QCommandLineOption roleOption(QStringLiteral("role"), - QStringLiteral("Daemon role (CoreRouter, CollectorSerial, CollectorCan, Collector, SqlServer)"), + QStringLiteral("Daemon role. CoreRouter: full daemon + D-Bus. " + "Collector* roles are legacy/in-process collectors; " + "prefer standalone softbus_edge_agent repo for Ethernet edge."), QStringLiteral("role"), QStringLiteral("CoreRouter")); parser.addOption(roleOption); @@ -65,7 +67,8 @@ int main(int argc, char *argv[]) role = CoreService::Role::SqlServer; } else { LOG_ERROR() << "Invalid role:" << roleString - << "valid roles are: CoreRouter, CollectorSerial, CollectorCan, Collector, SqlServer"; + << "valid roles are: CoreRouter, CollectorSerial, CollectorCan, Collector, SqlServer " + "(Ethernet edge uses separate softbus_edge_agent; see docs/EDGE_UPLINK_PROTOCOL_v1.md)"; return 1; } // 开启coreserive的初始化工作|线程 diff --git a/src/api/API_INTERFACE_TABLE.md b/src/api/API_INTERFACE_TABLE.md index 3e18aa0..d9f5ff8 100644 --- a/src/api/API_INTERFACE_TABLE.md +++ b/src/api/API_INTERFACE_TABLE.md @@ -76,7 +76,11 @@ | `get_device_tree` | 无 | `0` | `5003` | 返回 `tree` | | `list_devices` | 无 | `0` | - | 返回 `devices` | | `get_device` | `deviceId` | `0` | `4044` | 单设备 JSON；未找到 `device_not_found` | -| `patch_device_tree_node` | `id`, `patch`（字段同设备树节点） | `0` | `4045`/`5003` | 更新节点并协调；`node_not_found` | +| `patch_device_tree_node` | `id`, `patch`（字段同设备树节点） | `0` | `4045`/`5003` | 更新节点并协调；成功时可选带 `edgeTreeDownlink`（对 `params.edgeId` / `edges//` 作用域自动 `TREE_FULL`） | +| `list_edge_sessions` | 无 | `0` | - | 返回边缘 TCP 会话 `sessions`（`edgeId`/`helloOk`/`peer`） | +| `get_edge_tree_revision` | 无 | `0` | - | 返回 `treeRevision`（与边缘 `CTRL_TREE_PROPOSE` 乐观锁对齐） | +| `push_edge_tree` | `edgeId` | `0` | `4000`/`4047`/`5003`/`5005` | 向已 HELLO 的会话下行 `TREE_FULL`；成功返回 `sent`、`treeRevision`；无在线会话 `no_active_session`；ingress 未启用 `edge_ingress_disabled` | +| （edge TCP）`TREE_PROPOSE` | JSON：`edgeId`、`baseTreeRevision`、`entries` | - | `CONFLICT`/`FORBIDDEN_SCOPE` 等 | 与 D-Bus 共用 `DeviceTreeModel::applyEdgePropose`；`entries` 支持 `op` 省略/`patch` 与 **`op: upsert`**（边缘子树内新增或整节点合并），见 `docs/EDGE_UPLINK_PROTOCOL_v1.md` §6 | | `get_plugins` | 无 | `0` | - | 合并 protocol/framer/mapper 插件列表 | | `get_plugins_protocols` | 无 | `0` | - | `protocols` | | `get_plugins_mappers` | 无 | `0` | - | `mappers` | @@ -111,10 +115,12 @@ | `5002` | `persist_mappings_failed` | 映射规则内存更新成功但写入 `runtime_mappings.json` 失败 | | `5003` | `pipeline_unavailable` / `device_tree_unavailable` | Pipeline 或设备树模型未就绪 | | `5004` | `metadata_*` | 元数据字典读写/解析/保存失败 | +| `5005` | `edge_ingress_disabled` / `edge_ingress_unavailable` | 边缘 TCP ingress 未启用或未初始化 | | `4005` | `invalid_metadata` | `upsert_metadata` 缺少 `metadataId` | | `4044` | `device_not_found` | `get_device` 未找到 | | `4045` | `node_not_found` | `patch_device_tree_node` 未找到节点 | | `4046` | `metadata_not_found` | `get_metadata` / `delete_metadata` 未找到 | +| `4047` | `no_active_session` | `push_edge_tree` 时该 `edgeId` 无已 HELLO 的 TCP 会话 | ## 7) 维护建议 diff --git a/src/api/ipc_dbus/CommandDispatcher.cpp b/src/api/ipc_dbus/CommandDispatcher.cpp index 82106e2..3a3506b 100644 --- a/src/api/ipc_dbus/CommandDispatcher.cpp +++ b/src/api/ipc_dbus/CommandDispatcher.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include "core/metadata/MetadataRegistry.h" @@ -366,6 +367,45 @@ DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObj } return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("tree"), model->toDeviceTreeJson()}}}; } + if (action == QStringLiteral("list_edge_sessions")) { + const QJsonArray arr = softbus::device_bus::DeviceBus::instance().listEdgeIngressSessions(); + return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("sessions"), arr}}}; + } + if (action == QStringLiteral("get_edge_tree_revision")) { + return {0, + QStringLiteral("ok"), + QJsonObject{{QStringLiteral("treeRevision"), + softbus::device_bus::DeviceBus::instance().deviceTreeRevision()}}}; + } + if (action == QStringLiteral("push_edge_tree")) { + const QString edgeId = params.value(QStringLiteral("edgeId")).toString(); + if (edgeId.isEmpty()) { + return {4000, QStringLiteral("missing_edgeId"), {}}; + } + int sent = 0; + QString err; + const bool ok = softbus::device_bus::DeviceBus::instance().pushEdgeTreeFullDownlink(edgeId, &sent, &err); + if (!ok) { + int code = 5003; + if (err == QStringLiteral("no_active_session")) { + code = 4047; + } else if (err == QStringLiteral("missing_edgeId")) { + code = 4000; + } else if (err == QStringLiteral("edge_ingress_disabled") || err == QStringLiteral("edge_ingress_unavailable")) { + code = 5005; + } + return {code, err, QJsonObject{{QStringLiteral("edgeId"), edgeId}}}; + } + qint64 rev = 0; + if (auto model = softbus::device_bus::DeviceBus::instance().deviceTreeModel()) { + rev = model->treeRevision(); + } + return {0, + QStringLiteral("ok"), + QJsonObject{{QStringLiteral("edgeId"), edgeId}, + {QStringLiteral("sent"), sent}, + {QStringLiteral("treeRevision"), rev}}}; + } if (action == QStringLiteral("list_devices")) { const auto devices = softbus::device_bus::DeviceBus::instance().listDevices(); QJsonArray arr; @@ -389,42 +429,39 @@ DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObj if (!model) { return {5003, QStringLiteral("device_tree_unavailable"), {}}; } + QMutexLocker locker(&model->writeMutex()); const int idx = model->findIndexById(id); if (idx < 0) { return {4045, QStringLiteral("node_not_found"), {}}; } - DeviceTreeNode& node = model->nodesMutable()[idx]; - for (auto it = patch.constBegin(); it != patch.constEnd(); ++it) { - const QString k = it.key(); - if (k == QStringLiteral("params") && it.value().isObject()) { - QJsonObject p = node.params; - const QJsonObject patchParams = it.value().toObject(); - for (auto pit = patchParams.constBegin(); pit != patchParams.constEnd(); ++pit) { - const QString pk = pit.key(); - if (pk == QStringLiteral("stableKey") || pk == QStringLiteral("online") || pk == QStringLiteral("lastSeenTs")) { - continue; - } - p.insert(pk, pit.value()); - } - node.params = p; - } else if (k == QStringLiteral("name")) { - node.name = it.value().toString(); - } else if (k == QStringLiteral("type")) { - node.type = it.value().toString(); - } else if (k == QStringLiteral("endpoint")) { - node.endpoint = it.value().toString(); - } else if (k == QStringLiteral("driver")) { - node.driver = it.value().toString(); - } else if (k == QStringLiteral("capabilitiesRef")) { - node.capabilitiesRef = it.value().toString(); - } else if (k == QStringLiteral("enabled")) { - node.enabled = it.value().toBool(node.enabled); - } + if (!model->applyPatchToNodeById(id, patch)) { + return {4045, QStringLiteral("node_not_found"), {}}; } + const DeviceTreeNode& patchedNode = model->nodes().at(idx); + const QVector edgePushTargets = model->downlinkEdgeIdsForNode(patchedNode); model->rebuildIndex(); + model->bumpTreeRevision(); + locker.unlock(); softbus::device_bus::DeviceBus::instance().markDeviceTreeDirty(); (void)softbus::device_bus::DeviceBus::instance().reconcileFromTreeModel(); - return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("id"), id}}}; + QJsonObject data{{QStringLiteral("id"), id}}; + if (!edgePushTargets.isEmpty()) { + QJsonArray downlinkArr; + for (const QString& eid : edgePushTargets) { + int sent = 0; + QString err; + const bool ok = softbus::device_bus::DeviceBus::instance().pushEdgeTreeFullDownlink(eid, &sent, &err); + QJsonObject one{{QStringLiteral("edgeId"), eid}, + {QStringLiteral("ok"), ok}, + {QStringLiteral("sent"), sent}}; + if (!ok && !err.isEmpty()) { + one.insert(QStringLiteral("error"), err); + } + downlinkArr.append(one); + } + data.insert(QStringLiteral("edgeTreeDownlink"), downlinkArr); + } + return {0, QStringLiteral("ok"), data}; } if (action == QStringLiteral("get_plugins")) { const auto pm = softbus::device_bus::DeviceBus::instance().pluginManager(); diff --git a/src/core/CoreService.cpp b/src/core/CoreService.cpp index d6770b8..67e8ae5 100644 --- a/src/core/CoreService.cpp +++ b/src/core/CoreService.cpp @@ -68,6 +68,13 @@ bool CoreService::initialize() {QStringLiteral("ipcSocketName"), QStringLiteral("softbus_orchestrator_bus")}, {QStringLiteral("ipcOutboxPath"), QDir::homePath() + QStringLiteral("/softbus/run/dom_outbox.ndjson")} }}, + {QStringLiteral("edgeIngress"), QJsonObject{ + {QStringLiteral("enabled"), false}, + {QStringLiteral("listenAddress"), QStringLiteral("0.0.0.0")}, + {QStringLiteral("port"), 18765}, + {QStringLiteral("maxPayloadBytes"), 4 * 1024 * 1024}, + {QStringLiteral("authToken"), QString()} + }}, {QStringLiteral("plugins"), QJsonObject{}}, {QStringLiteral("device_tree"), QJsonArray{}} }; diff --git a/src/core/models/Types.h b/src/core/models/Types.h index d3abaa4..e22a4f0 100644 --- a/src/core/models/Types.h +++ b/src/core/models/Types.h @@ -9,7 +9,7 @@ namespace softbus::core::models enum class DataDomain : std::uint8_t { - PROCESS_PARAM = 0, // 过程参数 + PROCESS_PARAM = 0, // 过程参 数 CONTROL_CMD = 1, // 控制指令 MONITOR_DATA = 2, // 监控数据 MANAGE_DATA = 3, // 管理数据 diff --git a/src/device_bus/DeviceBus.cpp b/src/device_bus/DeviceBus.cpp index f1ed91e..a87ebad 100644 --- a/src/device_bus/DeviceBus.cpp +++ b/src/device_bus/DeviceBus.cpp @@ -2,6 +2,7 @@ #include #include +#include #include #include @@ -196,4 +197,31 @@ void DeviceBus::markDeviceTreeDirty() { d->deviceTreeDirty = true; } +QJsonArray DeviceBus::listEdgeIngressSessions() const { + if (!d || !d->manager) { + return {}; + } + return d->manager->listEdgeIngressSessions(); +} + +bool DeviceBus::pushEdgeTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut) const { + if (!d || !d->manager) { + if (sentCountOut) { + *sentCountOut = 0; + } + if (errorCodeOut) { + *errorCodeOut = QStringLiteral("device_bus_unavailable"); + } + return false; + } + return d->manager->pushEdgeTreeFullDownlink(edgeId, sentCountOut, errorCodeOut); +} + +qint64 DeviceBus::deviceTreeRevision() const { + if (!d || !d->treeModel) { + return 0; + } + return d->treeModel->treeRevision(); +} + } // namespace softbus::device_bus diff --git a/src/device_bus/DeviceBus.h b/src/device_bus/DeviceBus.h index 1b7c7ea..efa868f 100644 --- a/src/device_bus/DeviceBus.h +++ b/src/device_bus/DeviceBus.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -44,6 +45,12 @@ public: // 注册/更新设备树后调用，确保 shutdown 会落盘 void markDeviceTreeDirty(); + QJsonArray listEdgeIngressSessions() const; + qint64 deviceTreeRevision() const; + + /** 向已连接的 edge 会话下行 TREE_FULL；成功表示至少一路送达。 */ + bool pushEdgeTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut) const; + private: DeviceBus(); ~DeviceBus(); diff --git a/src/device_bus/manager/CanDeviceManager.cpp b/src/device_bus/manager/CanDeviceManager.cpp new file mode 100644 index 0000000..c2c5f5b --- /dev/null +++ b/src/device_bus/manager/CanDeviceManager.cpp @@ -0,0 +1,110 @@ +#include "device_bus/manager/CanDeviceManager.h" + +#include + +#include +#include + +#include "device_bus/registry/IDeviceRegistry.h" +#include "device_bus/tree/DeviceTreeModel.h" +#include "devices/DeviceTypes.h" +#include "utils/logs/logging.h" + +namespace softbus::device_bus::manager +{ + +namespace +{ + +static bool setIfChanged(QJsonObject& params, const QString& key, const QJsonValue& value) +{ + const auto it = params.find(key); + if (it == params.end() || it.value() != value) { + params.insert(key, value); + return true; + } + return false; +} + +static bool persistCanBitrate(DeviceTreeModel* treeModel, + const QString& endpoint, + int bitrate, + const std::function& dirtyCb) +{ + if (!treeModel || endpoint.isEmpty() || bitrate <= 0) { + return false; + } + const int idx = treeModel->findIndexByEndpoint(endpoint); + if (idx < 0) { + return false; + } + auto& nodes = treeModel->nodesMutable(); + if (idx >= static_cast(nodes.size())) { + return false; + } + auto& p = nodes[idx].params; + bool changed = setIfChanged(p, QStringLiteral("bitrate"), bitrate); + if (changed && dirtyCb) { + dirtyCb(); + } + return changed; +} + +} // namespace + +bool initializeCanDeviceWithDeps(const QString& endpoint, const QJsonObject& params, const CanInitDeps& deps) +{ + if (!deps.canDevicesByEndpoint || !deps.ingressAdapter || !deps.memoryPool) { + return false; + } + if (deps.canDevicesByEndpoint->contains(endpoint)) { + LOG_WARNING() << "DeviceBusManager: CAN device already exists for" << endpoint; + return false; + } + + int bitrate = params.value(QStringLiteral("bitrate")).toInt(500'000); + if (bitrate <= 0) { + bitrate = 500'000; + } + persistCanBitrate(deps.treeModel, endpoint, bitrate, deps.dirtyCb); + + QString iface = endpoint.trimmed(); + if (iface.startsWith(QStringLiteral("can:"), Qt::CaseInsensitive)) { + iface = iface.mid(4); + } + QString err; + std::unique_ptr bus(QCanBus::instance()->createDevice(QStringLiteral("socketcan"), iface, &err)); + if (!bus) { + LOG_ERROR() << "CanDeviceManager: createDevice failed" << endpoint << err; + return false; + } + + softbus::devices::DeviceInfo info; + info.id = -1; + info.endpoint = endpoint; + info.type = QStringLiteral("can"); + info.kind = softbus::devices::DeviceKind::Can; + info.protocol = params.value(QStringLiteral("protocol")).toString(QStringLiteral("can_raw")); + + softbus::devices::physical::CanDevice::Config cfg; + cfg.endpoint = endpoint; + cfg.bitrate = bitrate; + + QSharedPointer device( + new softbus::devices::physical::CanDevice(std::move(bus), cfg, info)); + device->bindIngress(deps.ingressAdapter, deps.memoryPool); + + if (deps.registry) { + deps.registry->registerDevice(info); + } + + if (!device->start()) { + LOG_ERROR() << "CanDeviceManager: connectDevice failed" << endpoint; + return false; + } + + deps.canDevicesByEndpoint->insert(endpoint, device); + return true; +} + +} // namespace softbus::device_bus::manager diff --git a/src/device_bus/manager/CanDeviceManager.h b/src/device_bus/manager/CanDeviceManager.h new file mode 100644 index 0000000..6fee088 --- /dev/null +++ b/src/device_bus/manager/CanDeviceManager.h @@ -0,0 +1,34 @@ +#pragma once + +#include +#include + +#include +#include +#include +#include + +#include "core/memory/MemoryPool.h" +#include "devices/physical/CanDevice.h" +#include "message_bus/ingress/IIngressPort.h" + +namespace softbus::device_bus::registry { class IDeviceRegistry; } + +class DeviceTreeModel; + +namespace softbus::device_bus::manager +{ + +struct CanInitDeps +{ + DeviceTreeModel* treeModel{nullptr}; + std::function dirtyCb; + softbus::device_bus::registry::IDeviceRegistry* registry{nullptr}; + softbus::message_bus::ingress::IIngressPort* ingressAdapter{nullptr}; + std::shared_ptr memoryPool; + QHash>* canDevicesByEndpoint{nullptr}; +}; + +bool initializeCanDeviceWithDeps(const QString& endpoint, const QJsonObject& params, const CanInitDeps& deps); + +} // namespace softbus::device_bus::manager diff --git a/src/device_bus/manager/DeviceBusManager.cpp b/src/device_bus/manager/DeviceBusManager.cpp index 41e568b..248a53d 100644 --- a/src/device_bus/manager/DeviceBusManager.cpp +++ b/src/device_bus/manager/DeviceBusManager.cpp @@ -15,12 +15,16 @@ #include "device_bus/DeviceBus.h" #include "message_bus/egress/EgressRouter.h" #include "message_bus/ingress/DriverIngressAdapter.h" +#include "message_bus/ingress/EdgeTcpIngressService.h" #include "message_bus/pipeline/PipelineEngine.h" #include "core/metadata/ProfileRegistry.h" #include "message_bus/pipeline/mapping/MappingRegistry.h" #include "message_bus/pipeline/rules/DynamicRuleRegistry.h" #include "core/plugin_system/PluginManager.h" #include "device_bus/manager/SerialDeviceManager.h" +#if SOFTBUS_DAEMON_HAVE_CAN +#include "device_bus/manager/CanDeviceManager.h" +#endif #include "devices/DeviceTypes.h" #include "utils/logs/logging.h" @@ -343,6 +347,28 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig) return false; } + const QJsonObject edgeIngressObj = rootConfig.value(QStringLiteral("edgeIngress")).toObject(); + softbus::message_bus::ingress::EdgeTcpIngressService::Config edgeCfg; + edgeCfg.enabled = edgeIngressObj.value(QStringLiteral("enabled")).toBool(false); + edgeCfg.listenAddress = edgeIngressObj.value(QStringLiteral("listenAddress")).toString(QStringLiteral("0.0.0.0")); + edgeCfg.port = edgeIngressObj.value(QStringLiteral("port")).toInt(18765); + edgeCfg.maxPayloadBytes = edgeIngressObj.value(QStringLiteral("maxPayloadBytes")).toInt(4 * 1024 * 1024); + edgeCfg.authToken = edgeIngressObj.value(QStringLiteral("authToken")).toString(); + m_edgeTcpIngress = std::make_unique(); + softbus::message_bus::ingress::EdgeTcpIngressService::Deps edgeDeps; + edgeDeps.ingress = m_ingressAdapter.get(); + edgeDeps.pool = m_memoryPool; + edgeDeps.tree = m_treeModel; + edgeDeps.markDeviceTreeDirty = [this]() { + if (m_dirtyCb) { + m_dirtyCb(); + } + }; + edgeDeps.reconcileFromTreeModel = [this]() { reconcileFromTreeModel(); }; + if (!m_edgeTcpIngress->start(edgeCfg, edgeDeps)) { + LOG_WARNING() << "DeviceBusManager: EdgeTcpIngressService start failed"; + } + const auto nodes = m_treeModel->nodes(); for (const auto& n : nodes) { if (!n.enabled) continue; @@ -355,6 +381,11 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig) if (kind == softbus::devices::DeviceKind::Serial) { initializeSerialDevice(n.endpoint, n.params); m_deviceEndpointsMap.insert(n.endpoint, kind); +#if SOFTBUS_DAEMON_HAVE_CAN + } else if (kind == softbus::devices::DeviceKind::Can) { + initializeCanDevice(n.endpoint, n.params); + m_deviceEndpointsMap.insert(n.endpoint, kind); +#endif } else { LOG_WARNING() << "DeviceBusManager: skip unsupported kind for endpoint=" << n.endpoint << "type=" << n.type; @@ -371,6 +402,9 @@ bool DeviceBusManager::reconcileFromTreeModel() } QSet desiredSerialEndpoints; +#if SOFTBUS_DAEMON_HAVE_CAN + QSet desiredCanEndpoints; +#endif const auto nodes = m_treeModel->nodes(); for (const auto& n : nodes) { if (!n.enabled) continue; @@ -387,6 +421,16 @@ bool DeviceBusManager::reconcileFromTreeModel() initializeSerialDevice(n.endpoint, n.params); } m_deviceEndpointsMap.insert(n.endpoint, kind); + } else if (kind == softbus::devices::DeviceKind::Can) { +#if SOFTBUS_DAEMON_HAVE_CAN + desiredCanEndpoints.insert(n.endpoint); + if (!m_canDevicesByEndpointSharedPtr.contains(n.endpoint)) { + initializeCanDevice(n.endpoint, n.params); + } + m_deviceEndpointsMap.insert(n.endpoint, kind); +#else + (void)n; +#endif } } @@ -398,22 +442,42 @@ bool DeviceBusManager::reconcileFromTreeModel() m_deviceEndpointsMap.remove(endpoint); } } +#if SOFTBUS_DAEMON_HAVE_CAN + const auto currentCan = m_canDevicesByEndpointSharedPtr.keys(); + for (const auto& endpoint : currentCan) { + if (!desiredCanEndpoints.contains(endpoint)) { + shutdownCanDevice(endpoint); + m_deviceEndpointsMap.remove(endpoint); + } + } +#endif return true; } void DeviceBusManager::shutdown() { + if (m_edgeTcpIngress) { + m_edgeTcpIngress->stop(); + m_edgeTcpIngress.reset(); + } + m_egressRunning.store(false); if (m_egressThread.joinable()) { m_egressThread.join(); } - // 关闭并销毁所有串口设备 + // 关闭并销毁所有串口与 CAN 设备 const auto keys = m_serialDevicesByEndpointSharedPtr.keys(); for (const auto& endpoint : keys) { shutdownSerialDevice(endpoint); } +#if SOFTBUS_DAEMON_HAVE_CAN + const auto canKeys = m_canDevicesByEndpointSharedPtr.keys(); + for (const auto& endpoint : canKeys) { + shutdownCanDevice(endpoint); + } +#endif if (m_pipelineEngine) { m_pipelineEngine->stop(); m_pipelineEngine->drain(); @@ -446,6 +510,28 @@ std::shared_ptr DeviceBusManager::p return m_pluginManager; } +QJsonArray DeviceBusManager::listEdgeIngressSessions() const +{ + if (!m_edgeTcpIngress) { + return {}; + } + return m_edgeTcpIngress->sessionsSnapshot(); +} + +bool DeviceBusManager::pushEdgeTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut) +{ + if (!m_edgeTcpIngress) { + if (sentCountOut) { + *sentCountOut = 0; + } + if (errorCodeOut) { + *errorCodeOut = QStringLiteral("edge_ingress_unavailable"); + } + return false; + } + return m_edgeTcpIngress->pushTreeFullDownlink(edgeId, sentCountOut, errorCodeOut); +} + void DeviceBusManager::initializeSerialDevice(const QString& endpoint, const QJsonObject& params) { softbus::device_bus::manager::SerialInitDeps deps; @@ -481,6 +567,37 @@ void DeviceBusManager::shutdownSerialDevice(const QString& endpoint) m_serialDevicesByEndpointSharedPtr.erase(it); } +#if SOFTBUS_DAEMON_HAVE_CAN +void DeviceBusManager::initializeCanDevice(const QString& endpoint, const QJsonObject& params) +{ + softbus::device_bus::manager::CanInitDeps deps; + deps.treeModel = m_treeModel.get(); + deps.dirtyCb = m_dirtyCb; + deps.registry = m_registry; + deps.ingressAdapter = m_ingressAdapter.get(); + deps.memoryPool = m_memoryPool; + deps.canDevicesByEndpoint = &m_canDevicesByEndpointSharedPtr; + (void)softbus::device_bus::manager::initializeCanDeviceWithDeps(endpoint, params, deps); +} + +void DeviceBusManager::shutdownCanDevice(const QString& endpoint) +{ + auto it = m_canDevicesByEndpointSharedPtr.find(endpoint); + if (it == m_canDevicesByEndpointSharedPtr.end()) { + return; + } + auto device = it.value(); + if (device) { + auto info = device->deviceInfo(); + if (m_registry && info.id >= 0) { + m_registry->unregisterDevice(info.id); + } + device->stop(); + } + m_canDevicesByEndpointSharedPtr.erase(it); +} +#endif + void DeviceBusManager::processEgressLoop() { while (m_egressRunning.load()) { diff --git a/src/device_bus/manager/DeviceBusManager.h b/src/device_bus/manager/DeviceBusManager.h index 9e17353..fbf23b3 100644 --- a/src/device_bus/manager/DeviceBusManager.h +++ b/src/device_bus/manager/DeviceBusManager.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -18,8 +19,12 @@ #include "device_bus/registry/IDeviceRegistry.h" #include "devices/DeviceTypes.h" #include "devices/physical/SerialDevice.h" +#if SOFTBUS_DAEMON_HAVE_CAN +#include "devices/physical/CanDevice.h" +#endif #include "message_bus/egress/EgressRouter.h" #include "message_bus/ingress/DriverIngressAdapter.h" +#include "message_bus/ingress/EdgeTcpIngressService.h" #include "message_bus/pipeline/PipelineEngine.h" class DeviceBusManager @@ -39,9 +44,16 @@ public: softbus::message_bus::pipeline::PipelineEngine* pipelineEngine() const; std::shared_ptr pluginManager() const; + QJsonArray listEdgeIngressSessions() const; + bool pushEdgeTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut); + private: void initializeSerialDevice(const QString& endpoint, const QJsonObject& params); void shutdownSerialDevice(const QString& endpoint); +#if SOFTBUS_DAEMON_HAVE_CAN + void initializeCanDevice(const QString& endpoint, const QJsonObject& params); + void shutdownCanDevice(const QString& endpoint); +#endif void processEgressLoop(); // void initializeCanDriver(); // void shutdownCanDriver(); @@ -58,6 +70,9 @@ private: QMap m_deviceEndpointsMap; QHash> m_serialDevicesByEndpointSharedPtr; +#if SOFTBUS_DAEMON_HAVE_CAN + QHash> m_canDevicesByEndpointSharedPtr; +#endif softbus::device_bus::registry::IDeviceRegistry* m_registry = nullptr; std::shared_ptr m_memoryPool; @@ -65,6 +80,7 @@ private: std::unique_ptr m_pluginHost; std::unique_ptr m_pipelineEngine; std::unique_ptr m_ingressAdapter; + std::unique_ptr m_edgeTcpIngress; std::unique_ptr m_egressRouter; std::atomic m_egressRunning{false}; std::thread m_egressThread; diff --git a/src/device_bus/tree/DeviceTreeModel.cpp b/src/device_bus/tree/DeviceTreeModel.cpp index bdcccc1..4471c25 100644 --- a/src/device_bus/tree/DeviceTreeModel.cpp +++ b/src/device_bus/tree/DeviceTreeModel.cpp @@ -1,10 +1,134 @@ #include "device_bus/tree/DeviceTreeModel.h" +#include + #include #include +#include #include "utils/logs/logging.h" +namespace +{ + +void applyJsonPatchToDeviceTreeNode(DeviceTreeNode& node, const QJsonObject& patch) +{ + for (auto it = patch.constBegin(); it != patch.constEnd(); ++it) { + const QString k = it.key(); + if (k == QStringLiteral("params") && it.value().isObject()) { + QJsonObject p = node.params; + const QJsonObject patchParams = it.value().toObject(); + for (auto pit = patchParams.constBegin(); pit != patchParams.constEnd(); ++pit) { + const QString pk = pit.key(); + if (pk == QStringLiteral("stableKey") || pk == QStringLiteral("online") || pk == QStringLiteral("lastSeenTs")) { + continue; + } + p.insert(pk, pit.value()); + } + node.params = p; + } else if (k == QStringLiteral("name")) { + node.name = it.value().toString(); + } else if (k == QStringLiteral("type")) { + node.type = it.value().toString(); + } else if (k == QStringLiteral("endpoint")) { + node.endpoint = it.value().toString(); + } else if (k == QStringLiteral("driver")) { + node.driver = it.value().toString(); + } else if (k == QStringLiteral("capabilitiesRef")) { + node.capabilitiesRef = it.value().toString(); + } else if (k == QStringLiteral("enabled")) { + node.enabled = it.value().toBool(node.enabled); + } + } +} + +bool idAllowedForEdgeProposeNew(const QString& id, const QString& edgeId) +{ + if (id.isEmpty() || edgeId.isEmpty()) { + return false; + } + const QString rootId = QStringLiteral("edges/%1").arg(edgeId); + const QString prefix = rootId + QLatin1Char('/'); + return (id == rootId) || id.startsWith(prefix); +} + +bool childrenIdsAllowedForEdge(const QVector& children, const QString& edgeId) +{ + const QString rootId = QStringLiteral("edges/%1").arg(edgeId); + const QString prefix = rootId + QLatin1Char('/'); + for (const QString& c : children) { + if (c.isEmpty()) { + return false; + } + if (!(c == rootId || c.startsWith(prefix))) { + return false; + } + } + return true; +} + +DeviceTreeNode deviceTreeNodeFromJsonObject(const QJsonObject& obj) +{ + DeviceTreeNode node; + node.id = obj.value(QStringLiteral("id")).toString(); + node.name = obj.value(QStringLiteral("name")).toString(); + node.type = obj.value(QStringLiteral("type")).toString(); + node.endpoint = obj.value(QStringLiteral("endpoint")).toString(); + node.driver = obj.value(QStringLiteral("driver")).toString(); + node.capabilitiesRef = obj.value(QStringLiteral("capabilitiesRef")).toString(); + node.params = obj.value(QStringLiteral("params")).toObject(); + node.enabled = obj.value(QStringLiteral("enabled")).toBool(true); + const QJsonValue ch = obj.value(QStringLiteral("children")); + if (ch.isArray()) { + for (const QJsonValue& cv : ch.toArray()) { + if (cv.isString()) { + node.childrenIds.push_back(cv.toString()); + } + } + } + return node; +} + +void mergeUpsertJsonIntoNode(DeviceTreeNode& node, const QJsonObject& src) +{ + if (src.contains(QStringLiteral("name"))) { + node.name = src.value(QStringLiteral("name")).toString(); + } + if (src.contains(QStringLiteral("type"))) { + node.type = src.value(QStringLiteral("type")).toString(); + } + if (src.contains(QStringLiteral("endpoint"))) { + node.endpoint = src.value(QStringLiteral("endpoint")).toString(); + } + if (src.contains(QStringLiteral("driver"))) { + node.driver = src.value(QStringLiteral("driver")).toString(); + } + if (src.contains(QStringLiteral("capabilitiesRef"))) { + node.capabilitiesRef = src.value(QStringLiteral("capabilitiesRef")).toString(); + } + if (src.contains(QStringLiteral("enabled"))) { + node.enabled = src.value(QStringLiteral("enabled")).toBool(node.enabled); + } + if (src.contains(QStringLiteral("params")) && src.value(QStringLiteral("params")).isObject()) { + QJsonObject p = node.params; + const QJsonObject patchParams = src.value(QStringLiteral("params")).toObject(); + for (auto pit = patchParams.constBegin(); pit != patchParams.constEnd(); ++pit) { + p.insert(pit.key(), pit.value()); + } + node.params = p; + } + if (src.contains(QStringLiteral("children")) && src.value(QStringLiteral("children")).isArray()) { + node.childrenIds.clear(); + for (const QJsonValue& cv : src.value(QStringLiteral("children")).toArray()) { + if (cv.isString()) { + node.childrenIds.push_back(cv.toString()); + } + } + } +} + +} // namespace + bool DeviceTreeModel::loadFromDaemonConfig(const QJsonObject& root, QString* errorMessage) { m_nodes.clear(); @@ -67,6 +191,9 @@ bool DeviceTreeModel::loadFromDaemonConfig(const QJsonObject& root, QString* err } LOG_INFO() << "DeviceTreeModel loaded" << m_nodes.size() << "nodes from daemon_config"; + m_treeRevision.store(std::max(qint64(1), + root.value(QStringLiteral("device_tree_revision")).toVariant().toLongLong()), + std::memory_order_relaxed); return true; } @@ -81,30 +208,51 @@ void DeviceTreeModel::rebuildIndex() } } +namespace +{ + +QJsonObject deviceNodeToJsonObject(const DeviceTreeNode& n) +{ + QJsonObject obj; + obj.insert(QStringLiteral("id"), n.id); + if (!n.name.isEmpty()) { + obj.insert(QStringLiteral("name"), n.name); + } + if (!n.type.isEmpty()) { + obj.insert(QStringLiteral("type"), n.type); + } + if (!n.endpoint.isEmpty()) { + obj.insert(QStringLiteral("endpoint"), n.endpoint); + } + if (!n.driver.isEmpty()) { + obj.insert(QStringLiteral("driver"), n.driver); + } + if (!n.capabilitiesRef.isEmpty()) { + obj.insert(QStringLiteral("capabilitiesRef"), n.capabilitiesRef); + } + if (!n.params.isEmpty()) { + obj.insert(QStringLiteral("params"), n.params); + } + obj.insert(QStringLiteral("enabled"), n.enabled); + + if (!n.childrenIds.isEmpty()) { + QJsonArray children; + for (const auto& c : n.childrenIds) { + children.append(c); + } + obj.insert(QStringLiteral("children"), children); + } + return obj; +} + +} // namespace + QJsonArray DeviceTreeModel::toDeviceTreeJson() const { QJsonArray arr; for (const auto& n : m_nodes) { - QJsonObject obj; - obj.insert(QStringLiteral("id"), n.id); - if (!n.name.isEmpty()) obj.insert(QStringLiteral("name"), n.name); - if (!n.type.isEmpty()) obj.insert(QStringLiteral("type"), n.type); - if (!n.endpoint.isEmpty()) obj.insert(QStringLiteral("endpoint"), n.endpoint); - if (!n.driver.isEmpty()) obj.insert(QStringLiteral("driver"), n.driver); - if (!n.capabilitiesRef.isEmpty()) obj.insert(QStringLiteral("capabilitiesRef"), n.capabilitiesRef); - if (!n.params.isEmpty()) obj.insert(QStringLiteral("params"), n.params); - obj.insert(QStringLiteral("enabled"), n.enabled); - - if (!n.childrenIds.isEmpty()) { - QJsonArray children; - for (const auto& c : n.childrenIds) { - children.append(c); - } - obj.insert(QStringLiteral("children"), children); - } - - arr.append(obj); + arr.append(deviceNodeToJsonObject(n)); } return arr; } @@ -225,3 +373,247 @@ bool DeviceTreeModel::markOfflineByStableKey(const QString& stableKey, const QSt return true; } +void DeviceTreeModel::bumpTreeRevision() +{ + m_treeRevision.fetch_add(1, std::memory_order_relaxed); +} + +bool DeviceTreeModel::applyPatchToNodeById(const QString& id, const QJsonObject& patch) +{ + const int idx = findIndexById(id); + if (idx < 0) { + return false; + } + applyJsonPatchToDeviceTreeNode(m_nodes[idx], patch); + return true; +} + +bool DeviceTreeModel::nodeBelongsToEdge(const DeviceTreeNode& node, const QString& edgeId) +{ + if (edgeId.isEmpty()) { + return false; + } + if (node.params.value(QStringLiteral("edgeId")).toString() == edgeId) { + return true; + } + const QString prefix = QStringLiteral("edges/%1/").arg(edgeId); + if (node.id == QStringLiteral("edges/%1").arg(edgeId) || node.id.startsWith(prefix)) { + return true; + } + return false; +} + +QJsonArray DeviceTreeModel::exportSubtreeForEdge(const QString& edgeId, qint64* outRevisionAtExport) const +{ + QJsonArray arr; + if (edgeId.isEmpty()) { + return arr; + } + QMutexLocker locker(&m_writeMutex); + if (outRevisionAtExport) { + *outRevisionAtExport = m_treeRevision.load(std::memory_order_relaxed); + } + for (const auto& n : m_nodes) { + if (nodeBelongsToEdge(n, edgeId)) { + arr.append(deviceNodeToJsonObject(n)); + } + } + return arr; +} + +QVector DeviceTreeModel::downlinkEdgeIdsForNode(const DeviceTreeNode& node) const +{ + QVector out; + auto pushUnique = [&out](const QString& s) { + const QString t = s.trimmed(); + if (t.isEmpty()) { + return; + } + for (const QString& x : out) { + if (x == t) { + return; + } + } + out.append(t); + }; + pushUnique(node.params.value(QStringLiteral("edgeId")).toString()); + static const QString pfx = QStringLiteral("edges/"); + if (node.id.startsWith(pfx)) { + const QString rest = node.id.mid(pfx.size()); + const int slash = rest.indexOf(QLatin1Char('/')); + pushUnique(slash < 0 ? rest : rest.left(slash)); + } + return out; +} + +DeviceTreeModel::EdgeProposeResult DeviceTreeModel::applyEdgePropose(const QString& edgeId, + qint64 baseRevision, + const QJsonArray& entries) +{ + EdgeProposeResult r; + QMutexLocker locker(&m_writeMutex); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + if (entries.isEmpty()) { + r.ok = false; + r.code = QStringLiteral("INVALID_JSON"); + r.message = QStringLiteral("entries must be non-empty"); + return r; + } + if (baseRevision != m_treeRevision.load(std::memory_order_relaxed)) { + r.ok = false; + r.code = QStringLiteral("CONFLICT"); + r.message = QStringLiteral("baseTreeRevision mismatch"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + + for (const QJsonValue& ev : entries) { + if (!ev.isObject()) { + r.ok = false; + r.code = QStringLiteral("INVALID_JSON"); + r.message = QStringLiteral("entries must be objects"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + const QJsonObject eo = ev.toObject(); + const QString id = eo.value(QStringLiteral("id")).toString(); + const QString op = eo.value(QStringLiteral("op")).toString(); + + if (op.isEmpty() || op == QStringLiteral("patch")) { + const QJsonObject patch = eo.value(QStringLiteral("patch")).toObject(); + if (id.isEmpty() || patch.isEmpty()) { + r.ok = false; + r.code = QStringLiteral("INVALID_JSON"); + r.message = QStringLiteral("each patch entry needs id and patch"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + const DeviceTreeNode* node = findById(id); + if (!node) { + r.ok = false; + r.code = QStringLiteral("NODE_NOT_FOUND"); + r.message = id; + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + if (!nodeBelongsToEdge(*node, edgeId)) { + r.ok = false; + r.code = QStringLiteral("FORBIDDEN_SCOPE"); + r.message = id; + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + if (!applyPatchToNodeById(id, patch)) { + r.ok = false; + r.code = QStringLiteral("NODE_NOT_FOUND"); + r.message = id; + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + } else if (op == QStringLiteral("upsert")) { + const QJsonObject nodeJo = eo.value(QStringLiteral("node")).toObject(); + if (id.isEmpty() || nodeJo.isEmpty()) { + r.ok = false; + r.code = QStringLiteral("INVALID_JSON"); + r.message = QStringLiteral("upsert entry needs id and non-empty node"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + if (nodeJo.contains(QStringLiteral("id"))) { + const QString nid = nodeJo.value(QStringLiteral("id")).toString(); + if (!nid.isEmpty() && nid != id) { + r.ok = false; + r.code = QStringLiteral("INVALID_JSON"); + r.message = QStringLiteral("node.id must match entry id"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + } + const int idx = findIndexById(id); + if (idx < 0) { + if (!idAllowedForEdgeProposeNew(id, edgeId)) { + r.ok = false; + r.code = QStringLiteral("FORBIDDEN_SCOPE"); + r.message = id; + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + DeviceTreeNode n = deviceTreeNodeFromJsonObject(nodeJo); + if (n.id.isEmpty()) { + n.id = id; + } + if (n.id != id) { + r.ok = false; + r.code = QStringLiteral("INVALID_JSON"); + r.message = QStringLiteral("upsert node id mismatch"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + if (!n.params.contains(QStringLiteral("edgeId"))) { + QJsonObject p = n.params; + p.insert(QStringLiteral("edgeId"), edgeId); + n.params = p; + } + if (!nodeBelongsToEdge(n, edgeId)) { + r.ok = false; + r.code = QStringLiteral("FORBIDDEN_SCOPE"); + r.message = id; + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + if (!childrenIdsAllowedForEdge(n.childrenIds, edgeId)) { + r.ok = false; + r.code = QStringLiteral("FORBIDDEN_SCOPE"); + r.message = QStringLiteral("children outside edge scope"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + m_nodes.push_back(n); + } else { + DeviceTreeNode& existing = m_nodes[idx]; + if (!nodeBelongsToEdge(existing, edgeId)) { + r.ok = false; + r.code = QStringLiteral("FORBIDDEN_SCOPE"); + r.message = id; + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + mergeUpsertJsonIntoNode(existing, nodeJo); + existing.id = id; + if (!existing.params.contains(QStringLiteral("edgeId"))) { + existing.params.insert(QStringLiteral("edgeId"), edgeId); + } + if (!nodeBelongsToEdge(existing, edgeId)) { + r.ok = false; + r.code = QStringLiteral("FORBIDDEN_SCOPE"); + r.message = id; + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + if (!childrenIdsAllowedForEdge(existing.childrenIds, edgeId)) { + r.ok = false; + r.code = QStringLiteral("FORBIDDEN_SCOPE"); + r.message = QStringLiteral("children outside edge scope"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + } + } else { + r.ok = false; + r.code = QStringLiteral("INVALID_JSON"); + r.message = QStringLiteral("unknown op"); + r.currentRevision = m_treeRevision.load(std::memory_order_relaxed); + return r; + } + rebuildIndex(); + } + + rebuildIndex(); + bumpTreeRevision(); + r.ok = true; + r.code = QStringLiteral("OK"); + r.newRevision = m_treeRevision.load(std::memory_order_relaxed); + r.currentRevision = r.newRevision; + return r; +} + diff --git a/src/device_bus/tree/DeviceTreeModel.h b/src/device_bus/tree/DeviceTreeModel.h index 3fa03c5..2c83fc2 100644 --- a/src/device_bus/tree/DeviceTreeModel.h +++ b/src/device_bus/tree/DeviceTreeModel.h @@ -5,6 +5,8 @@ #include #include #include +#include +#include // 轻量级设备树模型：从 daemon_config.json 的 device_tree 字段加载 @@ -53,8 +55,35 @@ public: // 当外部直接修改 nodesMutable() 后，需要重建索引 void rebuildIndex(); + // 与 D-Bus patch / 边缘 CTRL_TREE_PROPOSE 共用的写锁与 revision（见 docs/EDGE_UPLINK_PROTOCOL_v1.md） + QMutex& writeMutex() { return m_writeMutex; } + qint64 treeRevision() const { return m_treeRevision.load(std::memory_order_relaxed); } + void bumpTreeRevision(); + bool applyPatchToNodeById(const QString& id, const QJsonObject& patch); + + struct EdgeProposeResult + { + bool ok{false}; + QString code; // OK, CONFLICT, FORBIDDEN_SCOPE, NODE_NOT_FOUND, INVALID_JSON + QString message; + qint64 newRevision{0}; + qint64 currentRevision{0}; + }; + EdgeProposeResult applyEdgePropose(const QString& edgeId, qint64 baseRevision, const QJsonArray& entries); + + /** 导出属于该 edge 的节点子集（与 applyEdgePropose 作用域规则一致），用于下行 TREE_FULL。 + * 若传入 outRevisionAtExport，则写入导出时刻的 treeRevision（与 nodes 同一快照）。 */ + QJsonArray exportSubtreeForEdge(const QString& edgeId, qint64* outRevisionAtExport = nullptr) const; + + /** 节点若处于边缘作用域（params.edgeId 或 id 前缀 edges//...），返回应尝试下行 TREE_FULL 的 edgeId（去重）。 */ + QVector downlinkEdgeIdsForNode(const DeviceTreeNode& node) const; + private: + static bool nodeBelongsToEdge(const DeviceTreeNode& node, const QString& edgeId); + QVector m_nodes; QHash m_indexById; + mutable QMutex m_writeMutex; + std::atomic m_treeRevision{1}; }; diff --git a/src/devices/physical/CanDevice.cpp b/src/devices/physical/CanDevice.cpp new file mode 100644 index 0000000..4f29077 --- /dev/null +++ b/src/devices/physical/CanDevice.cpp @@ -0,0 +1,147 @@ +#include "devices/physical/CanDevice.h" + +#include + +#include +#include +#include + +#include "core/memory/MemoryPool.h" +#include "core/models/MessageRoutingUtils.h" +#include "core/models/RawBusMessage.h" +#include "message_bus/ingress/IIngressPort.h" +#include "utils/logs/logging.h" + +namespace softbus::devices::physical +{ + +static QString socketCanInterfaceName(const QString& endpoint) +{ + QString ep = endpoint.trimmed(); + if (ep.startsWith(QStringLiteral("can:"), Qt::CaseInsensitive)) { + return ep.mid(4); + } + return ep; +} + +CanDevice::CanDevice(std::unique_ptr bus, const Config& config, const softbus::devices::DeviceInfo& info) + : m_bus(std::move(bus)) + , m_config(config) + , m_info(info) +{ +} + +CanDevice::~CanDevice() +{ + stop(); +} + +bool CanDevice::start() +{ + if (!m_bus) { + LOG_ERROR() << "CanDevice: null bus"; + return false; + } + QObject::connect(m_bus.get(), &QCanBusDevice::framesReceived, m_bus.get(), [this]() { onFramesAvailable(); }); + QObject::connect(m_bus.get(), &QCanBusDevice::errorOccurred, m_bus.get(), [this](QCanBusDevice::CanBusError) { + onError(); + }); + + if (m_config.bitrate > 0) { + m_bus->setConfigurationParameter(QCanBusDevice::BitRateKey, m_config.bitrate); + } + LOG_INFO() << "CanDevice: connectDevice" << m_config.endpoint << "iface" << socketCanInterfaceName(m_config.endpoint); + return m_bus->connectDevice(); +} + +void CanDevice::stop() +{ + if (!m_bus) { + return; + } + QObject::disconnect(m_bus.get(), nullptr, m_bus.get(), nullptr); + if (m_bus->state() == QCanBusDevice::ConnectedState) { + m_bus->disconnectDevice(); + } + m_bus.reset(); +} + +softbus::devices::DeviceInfo CanDevice::deviceInfo() const +{ + return m_info; +} + +QString CanDevice::endpoint() const +{ + return m_config.endpoint; +} + +softbus::devices::DeviceKind CanDevice::kind() const +{ + return softbus::devices::DeviceKind::Can; +} + +void CanDevice::bindIngress(softbus::message_bus::ingress::IIngressPort* ingress, + std::shared_ptr pool) +{ + m_ingress = ingress; + m_pool = std::move(pool); +} + +void CanDevice::onFramesAvailable() +{ + if (!m_ingress || !m_pool || !m_bus) { + return; + } + while (m_bus->framesAvailable()) { + const QCanBusFrame frame = m_bus->readFrame(); + if (frame.frameType() != QCanBusFrame::DataFrame && frame.frameType() != QCanBusFrame::RemoteRequestFrame) { + continue; + } + const QByteArray busPayload = frame.payload(); + auto block = m_pool->allocate(static_cast(busPayload.size())); + if (!block || !block->data || block->capacity < static_cast(busPayload.size())) { + m_ingress->reportError(m_config.endpoint, QStringLiteral("memory pool exhausted")); + return; + } + if (!busPayload.isEmpty()) { + std::memcpy(block->data, busPayload.constData(), static_cast(busPayload.size())); + } + block->size = static_cast(busPayload.size()); + + softbus::core::models::RawBusMessage ctx; + ctx.direction = softbus::core::models::BusDirection::Upstream; + ctx.endpointHash = softbus::core::models::makeEndpointHash(m_config.endpoint); + ctx.protocol = softbus::core::models::protocolTypeFromHint(m_info.protocol); + if (ctx.protocol == softbus::core::models::ProtocolType::UNKNOWN) { + ctx.protocol = softbus::core::models::ProtocolType::CAN_RAW; + } + ctx.header.can.cobId = static_cast(frame.frameId() & 0x1FFFFFFFU); + ctx.header.can.isExtended = frame.hasExtendedFrameFormat(); + ctx.header.can.isRtr = (frame.frameType() == QCanBusFrame::RemoteRequestFrame); + ctx.logicalAddress = ctx.header.can.cobId & 0xFFU; + ctx.routingKey = softbus::core::models::makeRoutingKey(ctx.endpointHash, m_info.id); + ctx.timestampMs = QDateTime::currentMSecsSinceEpoch(); + ctx.deviceId = m_info.id; + ctx.payload.block = std::move(block); + ctx.payload.length = static_cast(busPayload.size()); + ctx.traceId = QStringLiteral("can:%1:%2:%3") + .arg(m_config.endpoint) + .arg(ctx.header.can.cobId) + .arg(ctx.timestampMs); + + if (!m_ingress->push(std::move(ctx))) { + m_ingress->reportError(m_config.endpoint, QStringLiteral("ingress queue full")); + } + } +} + +void CanDevice::onError() +{ + if (!m_ingress || !m_bus) { + return; + } + m_ingress->reportError(m_config.endpoint, m_bus->errorString()); +} + +} // namespace softbus::devices::physical diff --git a/src/devices/physical/CanDevice.h b/src/devices/physical/CanDevice.h new file mode 100644 index 0000000..9ea8c2d --- /dev/null +++ b/src/devices/physical/CanDevice.h @@ -0,0 +1,52 @@ +#pragma once + +#include + +#include + +#include "devices/DeviceTypes.h" +#include "devices/base/IPhysicalDevice.h" + +namespace softbus::message_bus::ingress { class IIngressPort; } +namespace softbus::core::memory { class MemoryPool; } + +class QCanBusDevice; + +namespace softbus::devices::physical +{ + +class CanDevice : public base::IPhysicalDevice +{ +public: + struct Config + { + QString endpoint; + int bitrate{500000}; + }; + + CanDevice(std::unique_ptr bus, const Config& config, const softbus::devices::DeviceInfo& info); + ~CanDevice() override; + + bool start() override; + void stop() override; + + softbus::devices::DeviceInfo deviceInfo() const override; + + QString endpoint() const; + softbus::devices::DeviceKind kind() const; + void bindIngress(softbus::message_bus::ingress::IIngressPort* ingress, + std::shared_ptr pool); + +private: + void onFramesAvailable(); + void onError(); + +private: + std::unique_ptr m_bus; + Config m_config; + softbus::devices::DeviceInfo m_info; + softbus::message_bus::ingress::IIngressPort* m_ingress{nullptr}; + std::shared_ptr m_pool; +}; + +} // namespace softbus::devices::physical diff --git a/src/message_bus/ingress/EdgeTcpIngressService.cpp b/src/message_bus/ingress/EdgeTcpIngressService.cpp new file mode 100644 index 0000000..769ccab --- /dev/null +++ b/src/message_bus/ingress/EdgeTcpIngressService.cpp @@ -0,0 +1,368 @@ +#include "message_bus/ingress/EdgeTcpIngressService.h" + +#include + +#include +#include +#include +#include +#include + +#include "message_bus/ingress/EdgeUplinkWire.h" +#include "utils/logs/logging.h" + +using softbus::message_bus::ingress::edge_uplink::FrameKind; +using softbus::message_bus::ingress::edge_uplink::decodeDataRawBusPayload; +using softbus::message_bus::ingress::edge_uplink::encodeOuterFrame; +using softbus::message_bus::ingress::edge_uplink::tryPopOneOuterFrame; + +namespace softbus::message_bus::ingress +{ + +EdgeTcpIngressService::EdgeTcpIngressService(QObject* parent) + : QObject(parent) +{ +} + +EdgeTcpIngressService::~EdgeTcpIngressService() +{ + stop(); +} + +bool EdgeTcpIngressService::start(const Config& cfg, const Deps& deps) +{ + m_cfg = cfg; + m_deps = deps; + if (!m_cfg.enabled) { + LOG_INFO() << "EdgeTcpIngressService: disabled in config (edgeIngress.enabled=false); no TCP listen. " + "Set edgeIngress.enabled=true in ~/softbus/config/daemon_config.json to accept edge_agent."; + return true; + } + if (!m_deps.ingress || !m_deps.pool || !m_deps.tree || !m_deps.markDeviceTreeDirty || !m_deps.reconcileFromTreeModel) { + LOG_ERROR() << "EdgeTcpIngressService: invalid deps"; + return false; + } + if (m_server) { + return true; + } + m_server = new QTcpServer(this); + connect(m_server, &QTcpServer::newConnection, this, &EdgeTcpIngressService::onNewConnection); + const QHostAddress addr(m_cfg.listenAddress); + if (!m_server->listen(addr, static_cast(m_cfg.port))) { + LOG_ERROR() << "EdgeTcpIngressService: listen failed" << m_cfg.listenAddress << m_cfg.port + << m_server->errorString(); + delete m_server; + m_server = nullptr; + return false; + } + LOG_INFO() << "EdgeTcpIngressService: listening" << m_cfg.listenAddress << "port" << m_cfg.port; + return true; +} + +void EdgeTcpIngressService::stop() +{ + if (m_server) { + m_server->close(); + for (auto it = m_clients.begin(); it != m_clients.end(); ++it) { + if (it->first) { + it->first->disconnect(this); + it->first->abort(); + } + } + m_clients.clear(); + m_server->deleteLater(); + m_server = nullptr; + } +} + +QJsonArray EdgeTcpIngressService::sessionsSnapshot() const +{ + QJsonArray arr; + for (const auto& kv : m_clients) { + if (!kv.second) { + continue; + } + QJsonObject o; + o.insert(QStringLiteral("edgeId"), kv.second->edgeId); + o.insert(QStringLiteral("helloOk"), kv.second->helloOk); + o.insert(QStringLiteral("peer"), + kv.first ? QStringLiteral("%1:%2").arg(kv.first->peerAddress().toString()).arg(kv.first->peerPort()) + : QString()); + arr.append(o); + } + return arr; +} + +void EdgeTcpIngressService::sendTreeFullToSocket(QTcpSocket* sock, const QString& edgeId) +{ + if (!sock || edgeId.isEmpty() || !m_deps.tree) { + return; + } + qint64 rev = 0; + const QJsonArray subtree = m_deps.tree->exportSubtreeForEdge(edgeId, &rev); + QJsonObject envelope; + envelope.insert(QStringLiteral("kind"), QStringLiteral("TREE_FULL")); + envelope.insert(QStringLiteral("edgeId"), edgeId); + envelope.insert(QStringLiteral("treeRevision"), rev); + envelope.insert(QStringLiteral("tree"), subtree); + const QByteArray pl = QJsonDocument(envelope).toJson(QJsonDocument::Compact); + const QByteArray frame = encodeOuterFrame(FrameKind::TreeFull, pl); + (void)sock->write(frame); +} + +bool EdgeTcpIngressService::pushTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut) +{ + if (sentCountOut) { + *sentCountOut = 0; + } + if (errorCodeOut) { + errorCodeOut->clear(); + } + if (!m_cfg.enabled) { + if (errorCodeOut) { + *errorCodeOut = QStringLiteral("edge_ingress_disabled"); + } + return false; + } + if (!m_deps.tree) { + if (errorCodeOut) { + *errorCodeOut = QStringLiteral("device_tree_unavailable"); + } + return false; + } + if (edgeId.isEmpty()) { + if (errorCodeOut) { + *errorCodeOut = QStringLiteral("missing_edgeId"); + } + return false; + } + int sent = 0; + for (auto& kv : m_clients) { + if (!kv.first || !kv.second || !kv.second->helloOk || kv.second->edgeId != edgeId) { + continue; + } + if (kv.first->state() != QAbstractSocket::ConnectedState) { + continue; + } + sendTreeFullToSocket(kv.first, edgeId); + ++sent; + } + if (sentCountOut) { + *sentCountOut = sent; + } + if (sent == 0) { + if (errorCodeOut) { + *errorCodeOut = QStringLiteral("no_active_session"); + } + return false; + } + return true; +} + +void EdgeTcpIngressService::onNewConnection() +{ + if (!m_server) { + return; + } + while (QTcpSocket* sock = m_server->nextPendingConnection()) { + auto ctx = std::make_unique(); + m_clients.emplace(sock, std::move(ctx)); + connect(sock, &QTcpSocket::readyRead, this, &EdgeTcpIngressService::onClientReadyRead); + connect(sock, &QTcpSocket::disconnected, this, &EdgeTcpIngressService::onClientDisconnected); + LOG_INFO() << "EdgeTcpIngressService: client connected" << sock->peerAddress().toString() << sock->peerPort(); + } +} + +void EdgeTcpIngressService::onClientDisconnected() +{ + auto* sock = qobject_cast(sender()); + if (!sock) { + return; + } + m_clients.erase(sock); + sock->deleteLater(); +} + +void EdgeTcpIngressService::onClientReadyRead() +{ + auto* sock = qobject_cast(sender()); + auto it = m_clients.find(sock); + if (it == m_clients.end() || !it->second) { + return; + } + ClientCtx& ctx = *it->second; + ctx.inbuf.append(sock->readAll()); + tryConsumeIncoming(sock, ctx); +} + +void EdgeTcpIngressService::tryConsumeIncoming(QTcpSocket* sock, ClientCtx& ctx) +{ + while (true) { + const auto popped = tryPopOneOuterFrame(ctx.inbuf, m_cfg.maxPayloadBytes); + if (!popped.has_value()) { + break; + } + const std::uint16_t kindU16 = static_cast(popped->kind); + handleFrame(sock, ctx, kindU16, popped->payload); + } +} + +void EdgeTcpIngressService::sendTreeResult(QTcpSocket* sock, const QJsonObject& body) +{ + if (!sock) { + return; + } + const QByteArray pl = QJsonDocument(body).toJson(QJsonDocument::Compact); + const QByteArray frame = encodeOuterFrame(FrameKind::TreeResult, pl); + (void)sock->write(frame); +} + +void EdgeTcpIngressService::handleFrame(QTcpSocket* sock, ClientCtx& ctx, std::uint16_t kindRaw, const QByteArray& payload) +{ + const auto kind = static_cast(kindRaw); + + if (kind == FrameKind::Hello) { + QJsonParseError pe; + const QJsonDocument doc = QJsonDocument::fromJson(payload, &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), QStringLiteral("INVALID_JSON")}, + {QStringLiteral("message"), pe.errorString()}}); + sock->disconnectFromHost(); + return; + } + const QJsonObject o = doc.object(); + ctx.edgeId = o.value(QStringLiteral("edgeId")).toString(); + if (ctx.edgeId.isEmpty()) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), QStringLiteral("INVALID_JSON")}, + {QStringLiteral("message"), QStringLiteral("missing edgeId")}}); + sock->disconnectFromHost(); + return; + } + if (!m_cfg.authToken.isEmpty()) { + if (o.value(QStringLiteral("authToken")).toString() != m_cfg.authToken) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), QStringLiteral("VALIDATION_ERROR")}, + {QStringLiteral("message"), QStringLiteral("bad authToken")}}); + sock->disconnectFromHost(); + return; + } + } + ctx.helloOk = true; + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), true}, + {QStringLiteral("code"), QStringLiteral("OK")}, + {QStringLiteral("message"), QStringLiteral("hello")}, + {QStringLiteral("currentTreeRevision"), m_deps.tree->treeRevision()}}); + sendTreeFullToSocket(sock, ctx.edgeId); + return; + } + + if (!ctx.helloOk) { + if (kind == FrameKind::DataRawBus && m_cfg.authToken.isEmpty()) { + // allow anonymous raw uplink when no authToken configured + } else if (kind == FrameKind::TreePropose) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), QStringLiteral("VALIDATION_ERROR")}, + {QStringLiteral("message"), QStringLiteral("HELLO required")}}); + return; + } else { + LOG_WARNING() << "EdgeTcpIngressService: HELLO required before this frame kind" << kindRaw; + sock->disconnectFromHost(); + return; + } + } + + if (kind == FrameKind::DataRawBus) { + QString err; + softbus::core::models::RawBusMessage ctxMsg; + QByteArray busPayload; + if (!decodeDataRawBusPayload(payload, ctxMsg, busPayload, &err)) { + LOG_WARNING() << "EdgeTcpIngressService: decode DataRawBus failed" << err; + return; + } + if (busPayload.isEmpty()) { + const auto p = ctxMsg.protocol; + if (p != softbus::core::models::ProtocolType::CAN_RAW + && p != softbus::core::models::ProtocolType::CAN_OPEN) { + LOG_WARNING() << "EdgeTcpIngressService: empty bus payload"; + return; + } + } + auto block = m_deps.pool->allocate(static_cast(busPayload.size())); + if (!block || !block->data || block->capacity < static_cast(busPayload.size())) { + LOG_WARNING() << "EdgeTcpIngressService: memory pool exhausted"; + return; + } + std::memcpy(block->data, busPayload.constData(), static_cast(busPayload.size())); + block->size = static_cast(busPayload.size()); + ctxMsg.payload.block = std::move(block); + ctxMsg.payload.length = static_cast(busPayload.size()); + if (!m_deps.ingress->push(std::move(ctxMsg))) { + m_deps.ingress->reportError(QStringLiteral("edge_tcp"), QStringLiteral("ingress queue full")); + } + return; + } + + if (kind == FrameKind::TreePropose) { + if (!ctx.helloOk) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), QStringLiteral("VALIDATION_ERROR")}, + {QStringLiteral("message"), QStringLiteral("HELLO required")}}); + return; + } + QJsonParseError pe; + const QJsonDocument doc = QJsonDocument::fromJson(payload, &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), QStringLiteral("INVALID_JSON")}, + {QStringLiteral("message"), pe.errorString()}}); + return; + } + const QJsonObject o = doc.object(); + const QString edgeId = o.value(QStringLiteral("edgeId")).toString(); + if (edgeId != ctx.edgeId) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), QStringLiteral("FORBIDDEN_SCOPE")}, + {QStringLiteral("message"), QStringLiteral("edgeId mismatch")}}); + return; + } + const qint64 baseRev = o.value(QStringLiteral("baseTreeRevision")).toVariant().toLongLong(); + const QJsonArray entries = o.value(QStringLiteral("entries")).toArray(); + const auto res = m_deps.tree->applyEdgePropose(edgeId, baseRev, entries); + if (!res.ok) { + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), false}, + {QStringLiteral("code"), res.code}, + {QStringLiteral("message"), res.message}, + {QStringLiteral("currentTreeRevision"), res.currentRevision}}); + return; + } + m_deps.markDeviceTreeDirty(); + m_deps.reconcileFromTreeModel(); + sendTreeResult(sock, + QJsonObject{{QStringLiteral("ok"), true}, + {QStringLiteral("code"), QStringLiteral("OK")}, + {QStringLiteral("newTreeRevision"), res.newRevision}, + {QStringLiteral("currentTreeRevision"), res.currentRevision}}); + int sentFull = 0; + (void)pushTreeFullDownlink(ctx.edgeId, &sentFull, nullptr); + return; + } + + if (kind == FrameKind::Ack || kind == FrameKind::Report) { + return; + } + + LOG_WARNING() << "EdgeTcpIngressService: unhandled frame kind" << kindRaw; +} + +} // namespace softbus::message_bus::ingress diff --git a/src/message_bus/ingress/EdgeTcpIngressService.h b/src/message_bus/ingress/EdgeTcpIngressService.h new file mode 100644 index 0000000..7f4de15 --- /dev/null +++ b/src/message_bus/ingress/EdgeTcpIngressService.h @@ -0,0 +1,80 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "core/memory/MemoryPool.h" +#include "device_bus/tree/DeviceTreeModel.h" +#include "message_bus/ingress/DriverIngressAdapter.h" + +namespace softbus::message_bus::ingress +{ + +class EdgeTcpIngressService final : public QObject +{ + Q_OBJECT + +public: + struct Config + { + bool enabled{false}; + QString listenAddress{QStringLiteral("0.0.0.0")}; + int port{18765}; + int maxPayloadBytes{4 * 1024 * 1024}; + QString authToken; // empty: first frame need not be HELLO auth (still recommend HELLO for edgeId) + }; + + struct Deps + { + DriverIngressAdapter* ingress{nullptr}; + std::shared_ptr pool; + std::shared_ptr tree; + std::function markDeviceTreeDirty; + std::function reconcileFromTreeModel; + }; + + explicit EdgeTcpIngressService(QObject* parent = nullptr); + ~EdgeTcpIngressService() override; + + bool start(const Config& cfg, const Deps& deps); + void stop(); + + QJsonArray sessionsSnapshot() const; + + /** 向所有已 HELLO 且 edgeId 匹配的会话发送 TREE_FULL（SBUP 外层 kind=TreeFull）。至少送达一路则返回 true。 */ + bool pushTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut); + +private slots: + void onNewConnection(); + void onClientReadyRead(); + void onClientDisconnected(); + +private: + struct ClientCtx + { + QByteArray inbuf; + QString edgeId; + bool helloOk{false}; + }; + + void handleFrame(QTcpSocket* sock, ClientCtx& ctx, std::uint16_t kindRaw, const QByteArray& payload); + void sendTreeResult(QTcpSocket* sock, const QJsonObject& body); + void sendTreeFullToSocket(QTcpSocket* sock, const QString& edgeId); + void tryConsumeIncoming(QTcpSocket* sock, ClientCtx& ctx); + + Config m_cfg; + Deps m_deps; + QTcpServer* m_server{nullptr}; + std::unordered_map> m_clients; +}; + +} // namespace softbus::message_bus::ingress diff --git a/src/message_bus/ingress/EdgeUplinkWire.cpp b/src/message_bus/ingress/EdgeUplinkWire.cpp new file mode 100644 index 0000000..ee9c243 --- /dev/null +++ b/src/message_bus/ingress/EdgeUplinkWire.cpp @@ -0,0 +1,140 @@ +#include "message_bus/ingress/EdgeUplinkWire.h" + +#include + +#include + +#include "utils/logs/logging.h" + +namespace softbus::message_bus::ingress::edge_uplink +{ + +static std::uint16_t readU16BE(const unsigned char* p) +{ + return qFromBigEndian(reinterpret_cast(p)); +} + +static std::uint32_t readU32BE(const unsigned char* p) +{ + return qFromBigEndian(reinterpret_cast(p)); +} + +static std::int32_t readI32BE(const unsigned char* p) +{ + return static_cast(readU32BE(p)); +} + +static std::int64_t readI64BE(const unsigned char* p) +{ + return qFromBigEndian(reinterpret_cast(p)); +} + +static std::uint64_t readU64BE(const unsigned char* p) +{ + return qFromBigEndian(reinterpret_cast(p)); +} + +QByteArray encodeOuterFrame(FrameKind kind, const QByteArray& payload) +{ + QByteArray out; + out.resize(static_cast(kOuterHeaderBytes + payload.size())); + unsigned char* d = reinterpret_cast(out.data()); + d[0] = static_cast(kMagic0); + d[1] = static_cast(kMagic1); + d[2] = static_cast(kMagic2); + d[3] = static_cast(kMagic3); + qToBigEndian(kWireVersion, d + 4); + qToBigEndian(static_cast(kind), d + 6); + qToBigEndian(static_cast(0), d + 8); + qToBigEndian(static_cast(payload.size()), d + 12); + if (!payload.isEmpty()) { + std::memcpy(d + kOuterHeaderBytes, payload.constData(), static_cast(payload.size())); + } + return out; +} + +std::optional tryPopOneOuterFrame(QByteArray& buffer, int maxPayloadBytes) +{ + if (buffer.size() < static_cast(kOuterHeaderBytes)) { + return std::nullopt; + } + const unsigned char* d = reinterpret_cast(buffer.constData()); + if (d[0] != static_cast(kMagic0) || d[1] != static_cast(kMagic1) + || d[2] != static_cast(kMagic2) || d[3] != static_cast(kMagic3)) { + LOG_WARNING() << "EdgeUplinkWire: bad magic, dropping byte"; + buffer.remove(0, 1); + return std::nullopt; + } + const std::uint16_t ver = readU16BE(d + 4); + if (ver != kWireVersion) { + LOG_WARNING() << "EdgeUplinkWire: unsupported wire_version" << ver; + buffer.clear(); + return std::nullopt; + } + const std::uint32_t payloadLen = readU32BE(d + 12); + if (maxPayloadBytes > 0 && payloadLen > static_cast(maxPayloadBytes)) { + LOG_WARNING() << "EdgeUplinkWire: payload_len exceeds max" << payloadLen << maxPayloadBytes; + buffer.clear(); + return std::nullopt; + } + const int total = static_cast(kOuterHeaderBytes) + static_cast(payloadLen); + if (payloadLen > 64u * 1024u * 1024u) { + LOG_WARNING() << "EdgeUplinkWire: absurd payload_len" << payloadLen; + buffer.clear(); + return std::nullopt; + } + if (buffer.size() < total) { + return std::nullopt; + } + OuterFramePop fr; + fr.kind = static_cast(readU16BE(d + 6)); + fr.payload = QByteArray(buffer.constData() + static_cast(kOuterHeaderBytes), static_cast(payloadLen)); + fr.consumedBytes = total; + buffer.remove(0, total); + return fr; +} + +bool decodeDataRawBusPayload(const QByteArray& payload, + softbus::core::models::RawBusMessage& ctx, + QByteArray& busPayloadOut, + QString* errorMessage) +{ + if (payload.size() < static_cast(kDataRawBusFixedBytes)) { + if (errorMessage) { + *errorMessage = QStringLiteral("data_rawbus_too_short"); + } + return false; + } + const unsigned char* p = reinterpret_cast(payload.constData()); + const std::uint16_t traceLen = readU16BE(p + 0); + if (traceLen > static_cast(kMaxTraceLen)) { + if (errorMessage) { + *errorMessage = QStringLiteral("trace_too_long"); + } + return false; + } + const int need = static_cast(kDataRawBusFixedBytes) + static_cast(traceLen); + if (payload.size() < need) { + if (errorMessage) { + *errorMessage = QStringLiteral("data_rawbus_truncated_trace"); + } + return false; + } + ctx = {}; + ctx.deviceId = readI32BE(p + 2); + ctx.endpointHash = readU32BE(p + 6); + ctx.protocol = static_cast(readU16BE(p + 10)); + ctx.logicalAddress = readU32BE(p + 14); + ctx.routingKey = readU64BE(p + 18); + std::memcpy(&ctx.header, p + 26, 8); + ctx.timestampMs = readI64BE(p + 34); + ctx.direction = static_cast(static_cast(p[42])); + if (traceLen > 0) { + ctx.traceId = QString::fromUtf8(reinterpret_cast(p + 48), static_cast(traceLen)); + } + busPayloadOut = payload.mid(need); + ctx.payload = {}; + return true; +} + +} // namespace softbus::message_bus::ingress::edge_uplink diff --git a/src/message_bus/ingress/EdgeUplinkWire.h b/src/message_bus/ingress/EdgeUplinkWire.h new file mode 100644 index 0000000..479398d --- /dev/null +++ b/src/message_bus/ingress/EdgeUplinkWire.h @@ -0,0 +1,53 @@ +#pragma once + +#include +#include + +#include +#include + +#include "core/models/RawBusMessage.h" + +namespace softbus::message_bus::ingress::edge_uplink +{ + +inline constexpr char kMagic0 = 'S'; +inline constexpr char kMagic1 = 'B'; +inline constexpr char kMagic2 = 'U'; +inline constexpr char kMagic3 = 'P'; +inline constexpr std::uint16_t kWireVersion = 1; + +enum class FrameKind : std::uint16_t +{ + Hello = 1, + Ack = 2, + DataRawBus = 3, + TreeFull = 4, + TreePatch = 5, + TreePropose = 6, + TreeResult = 7, + Report = 8, +}; + +inline constexpr std::size_t kOuterHeaderBytes = 16; +inline constexpr std::size_t kDataRawBusFixedBytes = 48; +inline constexpr int kMaxTraceLen = 512; + +struct OuterFramePop +{ + FrameKind kind{FrameKind::Hello}; + QByteArray payload; + int consumedBytes{0}; +}; + +std::optional tryPopOneOuterFrame(QByteArray& buffer, int maxPayloadBytes = 4 * 1024 * 1024); + +QByteArray encodeOuterFrame(FrameKind kind, const QByteArray& payload); + +// 填充除 payload.block 外的标量与 trace；busPayload 为总线原始字节，由调用方 allocate 后写入 ctx.payload +bool decodeDataRawBusPayload(const QByteArray& payload, + softbus::core::models::RawBusMessage& ctx, + QByteArray& busPayloadOut, + QString* errorMessage); + +} // namespace softbus::message_bus::ingress::edge_uplink diff --git a/tests/edge_uplink_wire_test.cpp b/tests/edge_uplink_wire_test.cpp new file mode 100644 index 0000000..8865bdd --- /dev/null +++ b/tests/edge_uplink_wire_test.cpp @@ -0,0 +1,57 @@ +#include +#include + +#include +#include + +#include "message_bus/ingress/EdgeUplinkWire.h" + +using softbus::message_bus::ingress::edge_uplink::FrameKind; +using softbus::message_bus::ingress::edge_uplink::encodeOuterFrame; +using softbus::message_bus::ingress::edge_uplink::tryPopOneOuterFrame; + +int main() +{ + const QByteArray inner = QByteArray("hello"); + const QByteArray frame = encodeOuterFrame(FrameKind::Hello, inner); + assert(frame.size() == 16 + inner.size()); + assert(frame.at(0) == 'S' && frame.at(1) == 'B' && frame.at(2) == 'U' && frame.at(3) == 'P'); + + QByteArray buf = frame; + const auto popped = tryPopOneOuterFrame(buf, 1024 * 1024); + assert(popped.has_value()); + assert(popped->kind == FrameKind::Hello); + assert(popped->payload == inner); + assert(buf.isEmpty()); + + // DATA_RAWBUS round-trip meta + QByteArray raw; + raw.resize(48 + 5 + 3); + unsigned char* w = reinterpret_cast(raw.data()); + qToBigEndian(static_cast(5), w + 0); // trace len + qToBigEndian(static_cast(7), reinterpret_cast(w + 2)); + qToBigEndian(static_cast(0xAABBCCDD), w + 6); + qToBigEndian(static_cast(0), w + 10); + qToBigEndian(static_cast(0), w + 12); + qToBigEndian(static_cast(1), w + 14); + qToBigEndian(static_cast(2), w + 18); + qToBigEndian(static_cast(0), w + 26); + qToBigEndian(static_cast(123456789LL), reinterpret_cast(w + 34)); + w[42] = static_cast(1); // direction + std::memset(w + 43, 0, 5); + std::memcpy(w + 48, "trace", 5); + std::memcpy(w + 53, "AB\x00", 3); + + softbus::core::models::RawBusMessage msg; + QByteArray busOut; + QString err; + const bool ok = softbus::message_bus::ingress::edge_uplink::decodeDataRawBusPayload(raw, msg, busOut, &err); + assert(ok); + assert(msg.deviceId == 7); + assert(msg.endpointHash == 0xAABBCCDDu); + assert(msg.traceId == QStringLiteral("trace")); + assert(busOut.size() == 3); + assert(busOut.at(0) == 'A'); + + return 0; +}