flowzl/softbus_daemon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

edge

Browse Source
This commit is contained in:
flower_linux
2026-05-19 16:42:01 +08:00
parent 238b814057
commit fcafa81db1
21 changed files with 224 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx
View File

Binary file not shown.

BIN
.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx
View File

Binary file not shown.

BIN
.cache/clangd/index/DeviceBus.h.170EB5C1B9D8C650.idx
View File

Binary file not shown.

BIN
.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx
View File

Binary file not shown.

BIN
.cache/clangd/index/DeviceBusManager.h.4DAD625CA48C7EE3.idx
View File

Binary file not shown.

BIN
.cache/clangd/index/DeviceTreeModel.cpp.D68EA11C8BBF4ADF.idx
View File

Binary file not shown.

BIN
.cache/clangd/index/DeviceTreeModel.h.A5637DBA59AB7F8A.idx
View File

Binary file not shown.

BIN
.cache/clangd/index/main.cpp.76A75E580B25A05E.idx
View File

Binary file not shown.

BIN
bin/edge_uplink_wire_test
View File

Binary file not shown.

BIN
bin/softbus_daemon
View File

Binary file not shown.

33
docs/EDGE_UPLINK_PROTOCOL_v1.md
View File

@@ -4,6 +4,7 @@
- **数据面**：`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`
- **调试面**：`DEBUG_TUNNEL`（双向 UTF-8 JSON，见 §10；须已 HELLO）
边缘仓库应维护与本文档 **字节级一致** 的副本（如 `docs/UPLINK_PROTOCOL_v1.md`）。
@@ -36,6 +37,7 @@
| 6 | TREE_PROPOSE | 边缘→核心 | UTF-8 JSON，见 §6 |
| 7 | TREE_RESULT | 核心→边缘 | UTF-8 JSON，见 §7 |
| 8 | REPORT | 边缘→核心 | UTF-8 JSON（预留） |
| 9 | DEBUG_TUNNEL | **双向** | UTF-8 JSON，见 §10；**仅**在会话已 HELLO 后允许 |
## 4. HELLO（边缘→核心）
@@ -144,3 +146,34 @@ UTF-8 JSON：
## 9. 与 orchestrator / D-Bus
核心侧设备树持久化与 `reconcile` 行为与现有 `patch_device_tree_node` 一致；本协议为 **额外入口**，不得绕过校验逻辑。
## 10. DEBUG_TUNNEL（`frame_kind = 9`，双向）
用于主机与边缘之间的 **调试指令与监控** 类 JSON，不替代 `DATA_RAWBUS` / `TREE_*` 控制面。
### 10.1 会话要求
- **两端**均须在 **HELLO 成功之后** 才能发送本帧；核心对未 HELLO 连接上的 `DEBUG_TUNNEL` 会断开（与 `TREE_FULL` 等下行一致）。
- 载荷为 **UTF-8 JSON 对象**（建议总大小远小于 `maxFrameBytes`）。
### 10.2 推荐信封（可选）
为便于对账与过滤，建议载荷使用统一形状（实现不强制解析，可原样记录日志）：
- `v`（integer）：信封版本，当前 **1**。
- `from`（string）：`"host"` 或 `"edge"`。
- `tsMs`（integer）：发送端单调时钟毫秒（如 `QDateTime::currentMSecsSinceEpoch()`）。
- `body`（object）：实际调试负载（指令、采样快照、订阅键等由产品约定）。
### 10.3 主机 → 边缘
- 核心通过 `executeCommand` 的 action **`edge_debug_send`** 向指定 `edgeId` 的已 HELLO 会话 **编码一帧** `DEBUG_TUNNEL` 并写入 TCP。
- `edge_debug_send` 的 `params`：**必填** `edgeId`；调试 JSON 可放在 **`debug` 对象**中，或省略 `debug` 时除 `edgeId` 外的其余 `params` 字段合并为 `body`；核心会再包一层 `v`/`from:"host"`/`tsMs`/`body` 后下行（见 `CommandDispatcher`）。
### 10.4 边缘 → 主机
- 边缘在已连接且 HELLO 后，将 JSON 对象编码为 `DEBUG_TUNNEL` 上行；核心在 `EdgeTcpIngressService` 中解析并 **打日志**（可扩展为内部总线/订阅）。
### 10.5 与边缘 Ops HTTP 的关系
- 边缘进程内 **`OpsUiHttp`** 提供本机 **`GET /debug_host`**（最近从主机收到的调试 JSON 环形缓冲）与 **`POST /debug_edge`**（将 body JSON 经 `DEBUG_TUNNEL` 发往核心），便于在现场用 `curl` 与核心侧 `edge_debug_send` 做 **闭环联调**。

11
plugins/protocols/modbus_rtu/framer/ModbusRtuFramerPlugin.cpp
View File

@@ -47,14 +47,15 @@ public:
m_lastActivity = m_lastRx;
if (m_len + n > m_cfg.maxAdu) {
// 功能 ： 如果数据块超过最大长度，则丢弃
LOG_WARNING() << "ModbusRtuFramerPlugin: overflow endpointHash=0x"
<< QString::number(m_endpointHash, 16);
<< QString::number(m_endpointHash, 16) << "bufLen=" << m_len << "chunk=" << n;
m_len = 0;
in.payload = {};
return;
}
std::memcpy(m_buf.data() + m_len, p, n);
m_len += n;
in.payload = {};
while (m_len > 0) {
auto pred = predictAduLength(m_buf.data(), m_len, m_cfg.maxAdu);
@@ -73,7 +74,9 @@ public:
auto blk = m_pool->allocate(*pred);
if (!blk || !blk->data || blk->capacity < *pred) {
return;
LOG_WARNING() << "ModbusRtuFramerPlugin: pool allocate failed endpointHash=0x"
<< QString::number(m_endpointHash, 16) << "aduLen=" << *pred << "bufLen=" << m_len;
break;
}
std::memcpy(blk->data, m_buf.data(), *pred);
blk->size = *pred;
@@ -104,8 +107,6 @@ public:
}
m_len = rest;
}
in.payload = {};
}
void reset() override { m_len = 0; }

1
src/api/API_INTERFACE_TABLE.md
View File

@@ -80,6 +80,7 @@
| `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_debug_send` | `edgeId`；可选 `debug`（object），若省略则将 `params` 去掉 `edgeId` 作为 body | `0` | `4000`/`4047`/`5003`/`5005` | 向已 HELLO 会话下行 **`DEBUG_TUNNEL`**（UTF-8 JSON）；成功返回 `sent`；载荷过大 `debug_payload_too_large`；见 `docs/EDGE_UPLINK_PROTOCOL_v1.md` §10 |
| （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` |

34
src/api/ipc_dbus/CommandDispatcher.cpp
View File

@@ -406,6 +406,40 @@ DispatchResult CommandDispatcher::dispatch(const QString& action, const QJsonObj
{QStringLiteral("sent"), sent},
{QStringLiteral("treeRevision"), rev}}};
}
if (action == QStringLiteral("edge_debug_send")) {
const QString edgeId = params.value(QStringLiteral("edgeId")).toString();
if (edgeId.isEmpty()) {
return {4000, QStringLiteral("missing_edgeId"), {}};
}
QJsonObject dbg = params.value(QStringLiteral("debug")).toObject();
if (dbg.isEmpty()) {
dbg = params;
dbg.remove(QStringLiteral("edgeId"));
}
QJsonObject envelope;
envelope.insert(QStringLiteral("v"), 1);
envelope.insert(QStringLiteral("from"), QStringLiteral("host"));
envelope.insert(QStringLiteral("tsMs"), QDateTime::currentMSecsSinceEpoch());
envelope.insert(QStringLiteral("body"), dbg);
int sent = 0;
QString err;
const bool ok = softbus::device_bus::DeviceBus::instance().pushEdgeDebugDownlink(edgeId, envelope, &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;
} else if (err == QStringLiteral("debug_payload_too_large")) {
code = 4000;
}
return {code, err, QJsonObject{{QStringLiteral("edgeId"), edgeId}}};
}
return {0, QStringLiteral("ok"), QJsonObject{{QStringLiteral("edgeId"), edgeId}, {QStringLiteral("sent"), sent}}};
}
if (action == QStringLiteral("list_devices")) {
const auto devices = softbus::device_bus::DeviceBus::instance().listDevices();
QJsonArray arr;

13
src/device_bus/DeviceBus.cpp
View File

@@ -217,6 +217,19 @@ bool DeviceBus::pushEdgeTreeFullDownlink(const QString& edgeId, int* sentCountOu
return d->manager->pushEdgeTreeFullDownlink(edgeId, sentCountOut, errorCodeOut);
}
bool DeviceBus::pushEdgeDebugDownlink(const QString& edgeId, const QJsonObject& payload, 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->pushEdgeDebugDownlink(edgeId, payload, sentCountOut, errorCodeOut);
}
qint64 DeviceBus::deviceTreeRevision() const {
if (!d || !d->treeModel) {
return 0;

3
src/device_bus/DeviceBus.h
View File

@@ -51,6 +51,9 @@ public:
/** 向已连接的 edge 会话下行 TREE_FULL；成功表示至少一路送达。 */
bool pushEdgeTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut) const;
/** 向边缘下行 `DebugTunnel`（JSON），用于主机侧调试/监控指令。 */
bool pushEdgeDebugDownlink(const QString& edgeId, const QJsonObject& payload, int* sentCountOut, QString* errorCodeOut) const;
private:
DeviceBus();
~DeviceBus();

98
src/device_bus/manager/DeviceBusManager.cpp
View File

@@ -347,6 +347,12 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig)
return false;
}
const QJsonObject deviceBusObj = rootConfig.value(QStringLiteral("deviceBus")).toObject();
m_localPhysicalDrivers = deviceBusObj.value(QStringLiteral("localPhysicalDrivers")).toBool(true);
if (!m_localPhysicalDrivers) {
LOG_INFO() << "DeviceBusManager: localPhysicalDrivers=false — 不在本机初始化串口/CAN；物理口由边缘承担。";
}
const QJsonObject edgeIngressObj = rootConfig.value(QStringLiteral("edgeIngress")).toObject();
softbus::message_bus::ingress::EdgeTcpIngressService::Config edgeCfg;
edgeCfg.enabled = edgeIngressObj.value(QStringLiteral("enabled")).toBool(false);
@@ -369,26 +375,28 @@ bool DeviceBusManager::initialize(const QJsonObject& rootConfig)
LOG_WARNING() << "DeviceBusManager: EdgeTcpIngressService start failed";
}
const auto nodes = m_treeModel->nodes();
for (const auto& n : nodes) {
if (!n.enabled) continue;
if (m_localPhysicalDrivers) {
const auto nodes = m_treeModel->nodes();
for (const auto& n : nodes) {
if (!n.enabled) continue;
auto kind = softbus::devices::kindFromType(n.type);
if (kind == softbus::devices::DeviceKind::Unknown) {
kind = softbus::devices::inferDeviceKindFromEndpoint(n.endpoint);
}
auto kind = softbus::devices::kindFromType(n.type);
if (kind == softbus::devices::DeviceKind::Unknown) {
kind = softbus::devices::inferDeviceKindFromEndpoint(n.endpoint);
}
if (kind == softbus::devices::DeviceKind::Serial) {
initializeSerialDevice(n.endpoint, n.params);
m_deviceEndpointsMap.insert(n.endpoint, kind);
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);
} 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;
} else {
LOG_WARNING() << "DeviceBusManager: skip unsupported kind for endpoint=" << n.endpoint
<< "type=" << n.type;
}
}
}
@@ -405,32 +413,34 @@ bool DeviceBusManager::reconcileFromTreeModel()
#if SOFTBUS_DAEMON_HAVE_CAN
QSet<QString> desiredCanEndpoints;
#endif
const auto nodes = m_treeModel->nodes();
for (const auto& n : nodes) {
if (!n.enabled) continue;
if (!n.params.value(QStringLiteral("online")).toBool(true)) continue;
if (m_localPhysicalDrivers) {
const auto nodes = m_treeModel->nodes();
for (const auto& n : nodes) {
if (!n.enabled) continue;
if (!n.params.value(QStringLiteral("online")).toBool(true)) continue;
auto kind = softbus::devices::kindFromType(n.type);
if (kind == softbus::devices::DeviceKind::Unknown) {
kind = softbus::devices::inferDeviceKindFromEndpoint(n.endpoint);
}
if (kind == softbus::devices::DeviceKind::Serial) {
desiredSerialEndpoints.insert(n.endpoint);
if (!m_serialDevicesByEndpointSharedPtr.contains(n.endpoint)) {
initializeSerialDevice(n.endpoint, n.params);
auto kind = softbus::devices::kindFromType(n.type);
if (kind == softbus::devices::DeviceKind::Unknown) {
kind = softbus::devices::inferDeviceKindFromEndpoint(n.endpoint);
}
m_deviceEndpointsMap.insert(n.endpoint, kind);
} else if (kind == softbus::devices::DeviceKind::Can) {
if (kind == softbus::devices::DeviceKind::Serial) {
desiredSerialEndpoints.insert(n.endpoint);
if (!m_serialDevicesByEndpointSharedPtr.contains(n.endpoint)) {
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);
desiredCanEndpoints.insert(n.endpoint);
if (!m_canDevicesByEndpointSharedPtr.contains(n.endpoint)) {
initializeCanDevice(n.endpoint, n.params);
}
m_deviceEndpointsMap.insert(n.endpoint, kind);
#else
(void)n;
(void)n;
#endif
}
}
}
@@ -532,6 +542,20 @@ bool DeviceBusManager::pushEdgeTreeFullDownlink(const QString& edgeId, int* sent
return m_edgeTcpIngress->pushTreeFullDownlink(edgeId, sentCountOut, errorCodeOut);
}
bool DeviceBusManager::pushEdgeDebugDownlink(const QString& edgeId, const QJsonObject& payload, int* sentCountOut, QString* errorCodeOut)
{
if (!m_edgeTcpIngress) {
if (sentCountOut) {
*sentCountOut = 0;
}
if (errorCodeOut) {
*errorCodeOut = QStringLiteral("edge_ingress_unavailable");
}
return false;
}
return m_edgeTcpIngress->pushDebugDownlink(edgeId, payload, sentCountOut, errorCodeOut);
}
void DeviceBusManager::initializeSerialDevice(const QString& endpoint, const QJsonObject& params)
{
softbus::device_bus::manager::SerialInitDeps deps;

4
src/device_bus/manager/DeviceBusManager.h
View File

@@ -46,6 +46,7 @@ public:
QJsonArray listEdgeIngressSessions() const;
bool pushEdgeTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut);
bool pushEdgeDebugDownlink(const QString& edgeId, const QJsonObject& payload, int* sentCountOut, QString* errorCodeOut);
private:
void initializeSerialDevice(const QString& endpoint, const QJsonObject& params);
@@ -86,6 +87,7 @@ private:
std::thread m_egressThread;
std::function<void()> m_dirtyCb;
/** false：不在本机打开 Serial/CAN；物理口由边缘承担，核心只收 DATA_RAWBUS。默认 true 兼容旧部署。 */
bool m_localPhysicalDrivers{true};
};

65
src/message_bus/ingress/EdgeTcpIngressService.cpp
View File

@@ -159,6 +159,58 @@ bool EdgeTcpIngressService::pushTreeFullDownlink(const QString& edgeId, int* sen
return true;
}
bool EdgeTcpIngressService::pushDebugDownlink(const QString& edgeId, const QJsonObject& payload, 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 (edgeId.isEmpty()) {
if (errorCodeOut) {
*errorCodeOut = QStringLiteral("missing_edgeId");
}
return false;
}
const QByteArray pl = QJsonDocument(payload).toJson(QJsonDocument::Compact);
if (pl.size() > m_cfg.maxPayloadBytes) {
if (errorCodeOut) {
*errorCodeOut = QStringLiteral("debug_payload_too_large");
}
return false;
}
const QByteArray frame = encodeOuterFrame(FrameKind::DebugTunnel, pl);
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;
}
if (kv.first->write(frame) == frame.size()) {
++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) {
@@ -358,6 +410,19 @@ void EdgeTcpIngressService::handleFrame(QTcpSocket* sock, ClientCtx& ctx, std::u
return;
}
if (kind == FrameKind::DebugTunnel) {
QJsonParseError pe;
const QJsonDocument doc = QJsonDocument::fromJson(payload, &pe);
if (pe.error != QJsonParseError::NoError || !doc.isObject()) {
LOG_WARNING() << "EdgeTcpIngressService: DebugTunnel invalid JSON" << pe.errorString();
return;
}
const QJsonObject o = doc.object();
LOG_INFO() << "EdgeDebugTunnel rx edgeId=" << ctx.edgeId
<< "payload=" << QJsonDocument(o).toJson(QJsonDocument::Compact);
return;
}
if (kind == FrameKind::Ack || kind == FrameKind::Report) {
return;
}

3
src/message_bus/ingress/EdgeTcpIngressService.h
View File

@@ -53,6 +53,9 @@ public:
/** 向所有已 HELLO 且 edgeId 匹配的会话发送 TREE_FULL（SBUP 外层 kind=TreeFull）。至少送达一路则返回 true。 */
bool pushTreeFullDownlink(const QString& edgeId, int* sentCountOut, QString* errorCodeOut);
/** 向已 HELLO 的会话发送 `DebugTunnel`（UTF-8 JSON），用于主机→边缘调试/监控指令。 */
bool pushDebugDownlink(const QString& edgeId, const QJsonObject& payload, int* sentCountOut, QString* errorCodeOut);
private slots:
void onNewConnection();
void onClientReadyRead();

2
src/message_bus/ingress/EdgeUplinkWire.h
View File

@@ -27,6 +27,8 @@ enum class FrameKind : std::uint16_t
TreePropose = 6,
TreeResult = 7,
Report = 8,
/** 双向 UTF-8 JSON 调试/监控（主机↔边缘，需已 HELLO）。见 docs/EDGE_UPLINK_PROTOCOL_v1.md §10。 */
DebugTunnel = 9,
};
inline constexpr std::size_t kOuterHeaderBytes = 16;