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

edge split

Browse Source
This commit is contained in:
flower_linux
2026-05-13 16:45:31 +08:00
parent df5eb5529c
commit 238b814057
31 changed files with 1934 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

368
src/message_bus/ingress/EdgeTcpIngressService.cpp Normal file
View File

@@ -0,0 +1,368 @@
#include "message_bus/ingress/EdgeTcpIngressService.h"
#include <cstring>
#include <QJsonArray>
#include <QJsonObject>
#include <QAbstractSocket>
#include <QJsonDocument>
#include <QJsonParseError>
#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<quint16>(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<ClientCtx>();
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<QTcpSocket*>(sender());
if (!sock) {
return;
}
m_clients.erase(sock);
sock->deleteLater();
}
void EdgeTcpIngressService::onClientReadyRead()
{
auto* sock = qobject_cast<QTcpSocket*>(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<std::uint16_t>(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<FrameKind>(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<std::size_t>(busPayload.size()));
if (!block || !block->data || block->capacity < static_cast<std::size_t>(busPayload.size())) {
LOG_WARNING() << "EdgeTcpIngressService: memory pool exhausted";
return;
}
std::memcpy(block->data, busPayload.constData(), static_cast<std::size_t>(busPayload.size()));
block->size = static_cast<std::size_t>(busPayload.size());
ctxMsg.payload.block = std::move(block);
ctxMsg.payload.length = static_cast<std::size_t>(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

80
src/message_bus/ingress/EdgeTcpIngressService.h Normal file
View File

@@ -0,0 +1,80 @@
#pragma once
#include <functional>
#include <memory>
#include <unordered_map>
#include <QByteArray>
#include <QJsonArray>
#include <QJsonObject>
#include <QObject>
#include <QString>
#include <QTcpServer>
#include <QTcpSocket>
#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<softbus::core::memory::MemoryPool> pool;
std::shared_ptr<DeviceTreeModel> tree;
std::function<void()> markDeviceTreeDirty;
std::function<void()> 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<QTcpSocket*, std::unique_ptr<ClientCtx>> m_clients;
};
} // namespace softbus::message_bus::ingress

140
src/message_bus/ingress/EdgeUplinkWire.cpp Normal file
View File

@@ -0,0 +1,140 @@
#include "message_bus/ingress/EdgeUplinkWire.h"
#include <cstring>
#include <QtEndian>
#include "utils/logs/logging.h"
namespace softbus::message_bus::ingress::edge_uplink
{
static std::uint16_t readU16BE(const unsigned char* p)
{
return qFromBigEndian<quint16>(reinterpret_cast<const quint8*>(p));
}
static std::uint32_t readU32BE(const unsigned char* p)
{
return qFromBigEndian<quint32>(reinterpret_cast<const quint8*>(p));
}
static std::int32_t readI32BE(const unsigned char* p)
{
return static_cast<std::int32_t>(readU32BE(p));
}
static std::int64_t readI64BE(const unsigned char* p)
{
return qFromBigEndian<qint64>(reinterpret_cast<const quint8*>(p));
}
static std::uint64_t readU64BE(const unsigned char* p)
{
return qFromBigEndian<quint64>(reinterpret_cast<const quint8*>(p));
}
QByteArray encodeOuterFrame(FrameKind kind, const QByteArray& payload)
{
QByteArray out;
out.resize(static_cast<int>(kOuterHeaderBytes + payload.size()));
unsigned char* d = reinterpret_cast<unsigned char*>(out.data());
d[0] = static_cast<unsigned char>(kMagic0);
d[1] = static_cast<unsigned char>(kMagic1);
d[2] = static_cast<unsigned char>(kMagic2);
d[3] = static_cast<unsigned char>(kMagic3);
qToBigEndian(kWireVersion, d + 4);
qToBigEndian(static_cast<std::uint16_t>(kind), d + 6);
qToBigEndian(static_cast<std::uint32_t>(0), d + 8);
qToBigEndian(static_cast<std::uint32_t>(payload.size()), d + 12);
if (!payload.isEmpty()) {
std::memcpy(d + kOuterHeaderBytes, payload.constData(), static_cast<std::size_t>(payload.size()));
}
return out;
}
std::optional<OuterFramePop> tryPopOneOuterFrame(QByteArray& buffer, int maxPayloadBytes)
{
if (buffer.size() < static_cast<int>(kOuterHeaderBytes)) {
return std::nullopt;
}
const unsigned char* d = reinterpret_cast<const unsigned char*>(buffer.constData());
if (d[0] != static_cast<unsigned char>(kMagic0) || d[1] != static_cast<unsigned char>(kMagic1)
|| d[2] != static_cast<unsigned char>(kMagic2) || d[3] != static_cast<unsigned char>(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<std::uint32_t>(maxPayloadBytes)) {
LOG_WARNING() << "EdgeUplinkWire: payload_len exceeds max" << payloadLen << maxPayloadBytes;
buffer.clear();
return std::nullopt;
}
const int total = static_cast<int>(kOuterHeaderBytes) + static_cast<int>(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<FrameKind>(readU16BE(d + 6));
fr.payload = QByteArray(buffer.constData() + static_cast<int>(kOuterHeaderBytes), static_cast<int>(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<int>(kDataRawBusFixedBytes)) {
if (errorMessage) {
*errorMessage = QStringLiteral("data_rawbus_too_short");
}
return false;
}
const unsigned char* p = reinterpret_cast<const unsigned char*>(payload.constData());
const std::uint16_t traceLen = readU16BE(p + 0);
if (traceLen > static_cast<std::uint16_t>(kMaxTraceLen)) {
if (errorMessage) {
*errorMessage = QStringLiteral("trace_too_long");
}
return false;
}
const int need = static_cast<int>(kDataRawBusFixedBytes) + static_cast<int>(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<softbus::core::models::ProtocolType>(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<softbus::core::models::BusDirection>(static_cast<signed char>(p[42]));
if (traceLen > 0) {
ctx.traceId = QString::fromUtf8(reinterpret_cast<const char*>(p + 48), static_cast<int>(traceLen));
}
busPayloadOut = payload.mid(need);
ctx.payload = {};
return true;
}
} // namespace softbus::message_bus::ingress::edge_uplink

53
src/message_bus/ingress/EdgeUplinkWire.h Normal file
View File

@@ -0,0 +1,53 @@
#pragma once
#include <cstdint>
#include <optional>
#include <QByteArray>
#include <QString>
#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<OuterFramePop> 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