diff --git a/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx b/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx index 057b388..6ac8de8 100644 Binary files a/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx and b/.cache/clangd/index/CoreService.cpp.4E2160035C5FBDAD.idx differ diff --git a/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx b/.cache/clangd/index/DeviceBus.cpp.3FE983A42FE7A6A8.idx index 7233d24..3d4ccdd 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 9192706..92dace8 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.cpp.134168DDD210A4ED.idx b/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx index 966eddf..f58f38e 100644 Binary files a/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx and b/.cache/clangd/index/DeviceBusManager.cpp.134168DDD210A4ED.idx differ diff --git a/.cache/clangd/index/DeviceMonitorService.cpp.3C0446CEC6FFCFAC.idx b/.cache/clangd/index/DeviceMonitorService.cpp.3C0446CEC6FFCFAC.idx index ec900ee..fdde235 100644 Binary files a/.cache/clangd/index/DeviceMonitorService.cpp.3C0446CEC6FFCFAC.idx and b/.cache/clangd/index/DeviceMonitorService.cpp.3C0446CEC6FFCFAC.idx differ diff --git a/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx b/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx index d10ad8f..db8776b 100644 Binary files a/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx and b/.cache/clangd/index/SerialDevice.cpp.E5FA00DC79EFDCA9.idx differ diff --git a/.cache/clangd/index/UdevDeviceMonitor.cpp.8FD4DE713A589E66.idx b/.cache/clangd/index/UdevDeviceMonitor.cpp.8FD4DE713A589E66.idx index 683885e..e5bf318 100644 Binary files a/.cache/clangd/index/UdevDeviceMonitor.cpp.8FD4DE713A589E66.idx and b/.cache/clangd/index/UdevDeviceMonitor.cpp.8FD4DE713A589E66.idx differ diff --git a/.cache/clangd/index/main.cpp.76A75E580B25A05E.idx b/.cache/clangd/index/main.cpp.76A75E580B25A05E.idx index 99e0f84..0236563 100644 Binary files a/.cache/clangd/index/main.cpp.76A75E580B25A05E.idx and b/.cache/clangd/index/main.cpp.76A75E580B25A05E.idx differ diff --git a/CMakeLists.txt b/CMakeLists.txt index f29aca5..4683010 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -30,7 +30,8 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_AUTORCC ON) set(CMAKE_AUTOMOC ON) # 查找Qt6（守护进程只需要 Core/DBus/Network/SerialPort/Sql） -find_package(Qt6 6.5 REQUIRED COMPONENTS Core SerialPort Network Sql DBus) +find_package(Qt6 6.5 REQUIRED COMPONENTS Core SerialPort Network Sql DBus HttpServer) + include(CheckIncludeFileCXX) # 检测编译平台 @@ -62,6 +63,14 @@ qt_add_executable(softbus_daemon src/api/ipc_dbus/IpcDBus.cpp src/api/ipc_dbus/CommandDispatcher.h src/api/ipc_dbus/CommandDispatcher.cpp + src/api/http/RestApiServer.h + src/api/http/RestApiServer.cpp + src/api/http/RestApiHttpConfig.h + src/api/http/RestApiHttpConfig.cpp + src/api/http/RestApiCommandHttp.h + src/api/http/RestApiCommandHttp.cpp + src/api/http/RestApiRoutes.h + src/api/http/RestApiRoutes.cpp # device types src/devices/DeviceTypes.h @@ -125,6 +134,7 @@ qt_add_executable(softbus_daemon src/core/plugin_system/PluginManager.cpp src/core/plugin_system/PluginHost.h src/core/plugin_system/PluginHost.cpp + src/core/models/RawBusMessage.h plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp plugins/protocols/modbus_rtu/mapper/ModbusRtuProtocolMapperPlugin.h @@ -187,6 +197,7 @@ target_link_libraries(softbus_daemon Qt6::DBus Qt6::SerialPort Qt6::Network + Qt6::HttpServer ) # Platform-specific deps diff --git a/bin/softbus_daemon b/bin/softbus_daemon index 4131c63..17d8ea2 100755 Binary files a/bin/softbus_daemon and b/bin/softbus_daemon differ diff --git a/docs/数据流转.md b/docs/数据流转.md new file mode 100644 index 0000000..2cfb21d --- /dev/null +++ b/docs/数据流转.md @@ -0,0 +1,145 @@ +# 数据流转 + +## 1. 概念约定 + +- **物理设备**：真实硬件端口（当前主要是串口），由 `SerialQtDriver` 直接读写。 +- **设备实例**：`SerialDevice`，负责单设备生命周期、驱动回调、上下行数据接入。 +- **原始总线消息**：`RawBusMessage`，包含 `endpoint`、`protocolHint`、`payload`、`traceId`、`direction` 等基础字段。 +- **流水线上下文**：`PipelineContext`，继承 `RawBusMessage`，增加 `frameKind`、`frameSeq` 等管道字段。 +- **业务消息**：经过协议解析和映射后得到 `DOMMessage`（统一语义数据）。 + +--- + +## 2. 上行主链路（串口 -> 消息总线） + +### 2.1 设备初始化 + +`SerialDeviceManager` 完成以下动作： + +1. 根据配置创建并配置 `SerialQtDriver`（波特率、校验、停止位、流控）。 +2. 创建 `SerialDevice` 并注入 `DeviceInfo`（含协议提示，如 `modbus_rtu`）。 +3. 通过 `bindIngress(...)` 绑定： + - `IIngressPort`（当前实现为 `DriverIngressAdapter`） + - `MemoryPool`（用于 payload 内存块分配） +4. 调用 `start()` 打开端口并注册读回调/错误回调。 + +### 2.2 读回调进入 ingress + +当串口收到字节流时，`SerialDevice::onRead(...)` 执行： + +1. 从 `MemoryPool` 分配内存块并拷贝原始字节。 +2. 构造 `RawBusMessage`： + - `direction = Upstream` + - `endpoint = 串口名` + - `protocolHint = DeviceInfo.protocol` + - `payload/payloadSize = 原始字节数据` + - `traceId = serial::` +3. 调用 `m_ingress->push(std::move(ctx))` 进入统一入口。 +4. 若内存池不足或 ingress 队列满，通过 `reportError(...)` 记录错误。 + +### 2.3 ingress 适配为 PipelineContext + +`DriverIngressAdapter::push(...)` 负责轻量转换： + +1. 新建 `PipelineContext`。 +2. `pipelineCtx.raw() = std::move(ctx)` 将原始消息搬运进管道上下文。 +3. 调用 `PipelineEngine::enqueueIngress(...)` 入总线处理队列。 + +--- + +## 3. PipelineEngine 内部流转 + +### 3.1 入口分流 + +`enqueueIngress(...)` 分两种模式： + +- **并行模式（parallelPipeline=true）** + - 按 `endpoint` 创建/复用独立 `EndpointFramingState`（每端点一个分帧线程 + chunk 队列）。 + - chunk 先进入端点队列，端点线程负责分帧并赋值 `frameSeq`。 + - 完整帧进入 `m_framedQ`，由解析线程池处理。 +- **单线程模式（parallelPipeline=false）** + - 直接进入 `m_ingressQ`，由单 worker 串行处理。 + +### 3.2 Stage 1：分帧（Framer） + +`dispatchFramer(...)` 的选择策略： + +1. 优先根据 `protocolHint` 从插件管理器选择协议分帧插件。 +2. 按 `endpoint + pluginId` 复用 framer session（保持协议状态机连续性）。 +3. 若没有可用插件，降级到 `PassthroughFramer`（透传/整包策略）。 + +输出：`frameKind = CompleteFrame` 的帧批次。 + +### 3.3 Stage 2/3/4：解析 -> 过滤 -> 校验 + +`runStages234(...)` 核心流程： + +1. **Stage2 Parser** + - 选择 `IProtocolPlugin`，复用 parser session。 + - `parse(rawCtx, envelope)` 产出协议封装 `ProtocolEnvelope`。 + - 选择 mapper 插件并映射为 `std::vector`。 +2. **Stage3 Filter** + - 对 `DOMMessage` 执行过滤规则（`ProtocolParseFilter::process`）。 +3. **Stage4 Validator** + - 最低校验当前为 `messageId` 非空。 +4. **Transform/Enrich** + - 应用动态规则（scale/offset/unit）。 + - 写入 trace，补充富化字段。 + +说明：当前 `runStages234` 只返回布尔结果，`EnrichedMessage` 由内部生成，后续可接持久化/发布环节。 + +### 3.4 出口与有序保证 + +- 未开启有序出口：解析成功后直接进入 `m_egressQ`。 +- 开启 `orderedEgress`： + - 解析线程先写入 `m_mergeQ`。 + - `mergerLoop` 按 `endpoint + frameSeq` 重排。 + - 使用 `nextExpected` 保证同端点有序输出。 + - 超过 `maxReorderDepth` 的乱序缓存上限会丢弃并计数。 + +--- + +## 4. 错误与背压点 + +主要失败点及表现： + +1. **内存池分配失败**：`SerialDevice::onRead` 直接 `reportError(memory pool exhausted)`。 +2. **入口队列满**：`enqueueIngress` 返回 false，计入 drop。 +3. **分帧/解析插件缺失或失败**：对应帧在阶段中被丢弃。 +4. **过滤/校验失败**：消息跳过，不进入下游。 +5. **重排深度超限**：`mergeDrop` 增加并告警。 + +--- + +## 5. 当前链路时序（上行） + +```text +SerialQtDriver(read callback) + -> SerialDevice::onRead + -> MemoryPool::allocate + RawBusMessage + -> IIngressPort::push (DriverIngressAdapter) + -> PipelineEngine::enqueueIngress + -> [Framer] dispatchFramer/feedChunk + -> [Stage2] parse + map -> DOMMessage + -> [Stage3] filter + -> [Stage4] validate + -> transform/enrich + -> egressQ (有序模式下先 merge 再 egress) +``` + +--- + +## 6. 下行（现状） + +- `RawBusMessage.direction` 已支持 `Downstream` 语义。 +- `SerialDevice` 已提供 `writePayload(...)` 写入能力。 +- 当前代码中 egress 路由到物理设备的绑定仍预留（`SerialDeviceManager` 里相关逻辑已注释），后续可补齐： + - egress 消息 -> `EgressRouter` -> endpoint 设备 -> `writePayload`。 + +--- + +## 7. 维护建议 + +1. `traceId` 建议在全链路透传到最终发布侧，便于定位帧级问题。 +2. 为 Stage2/Stage3/Stage4 增加细粒度指标（成功率、耗时、drop 原因）。 +3. 文档后续可新增“协议插件开发约定”章节（framer/parser/mapper 三类插件接口对照）。 \ No newline at end of file diff --git a/main.cpp b/main.cpp index 3b8dea4..e362910 100644 --- a/main.cpp +++ b/main.cpp @@ -1,149 +1,159 @@ -#include -#include -#include -#include -#include -#include "src/utils/logs/logging.h" -#include "src/core/CoreService.h" -#include "src/api/ipc_dbus/IpcDBus.h" - -void signalHandler(int signal) -{ - LOG_INFO() << "Received signal:" << signal; - CoreService* coreService = CoreService::instance(); - if (coreService) { - coreService->shutdown(); - } - - QCoreApplication::quit(); -} - -int main(int argc, char *argv[]) -{ - // ================================== - // 创建并配置 Qt 应用程序对象 - // ================================== - QCoreApplication app(argc, argv); - - app.setApplicationName("soft_bus_daemon"); - app.setApplicationVersion("1.1.0"); - app.setOrganizationName("soft_bus"); - - // ================================== - // 设置 Qt 应用程序信息 - // ================================== - QCommandLineParser parser; - parser.setApplicationDescription(QStringLiteral("Soft Bus Daemon")); - parser.addHelpOption(); - parser.addVersionOption(); - QCommandLineOption roleOption(QStringLiteral("role"), - QStringLiteral("Daemon role (CoreRouter, CollectorSerial, CollectorCan, Collector, SqlServer)"), - QStringLiteral("role"), - QStringLiteral("CoreRouter")); - parser.addOption(roleOption); - QCommandLineOption dbusBusOption(QStringLiteral("dbus-bus"), - QStringLiteral("DBus bus to use: system|session|auto (default: auto)"), - QStringLiteral("bus"), - QStringLiteral("auto")); - parser.addOption(dbusBusOption); - parser.process(app); - - - const QString roleString = parser.value(roleOption).trimmed(); - const QString dbusBus = parser.value(dbusBusOption).trimmed(); - CoreService::Role role = CoreService::Role::CoreRouter; // 默认角色 - if (roleString == QStringLiteral("CoreRouter")) { - role = CoreService::Role::CoreRouter; - } else if (roleString == QStringLiteral("CollectorSerial")) { - role = CoreService::Role::CollectorSerial; - } else if (roleString == QStringLiteral("CollectorCan")) { - role = CoreService::Role::CollectorCan; - } else if (roleString == QStringLiteral("Collector")) { - role = CoreService::Role::Collector; - } else if (roleString == QStringLiteral("SqlServer")) { - role = CoreService::Role::SqlServer; - } else { - LOG_ERROR() << "Invalid role:" << roleString - << "valid roles are: CoreRouter, CollectorSerial, CollectorCan, Collector, SqlServer"; - return 1; - } - // 开启coreserive的初始化工作|线程 - // QThread* coreServiceThread = new QThread(); - // CoreService* coreService = new CoreService(); - // coreService->moveToThread(coreServiceThread); - // QObject::connect(coreServiceThread, &QThread::started, coreService, &CoreService::initialize); - // QObject::connect(coreServiceThread, &QThread::finished, coreService, &CoreService::deleteLater); - // QObject::connect(coreServiceThread, &QThread::finished, coreServiceThread, &QThread::deleteLater); - // coreServiceThread->start(); - - CoreService* coreService = CoreService::instance(); - if (!coreService->initialize()) { - LOG_ERROR() << "CoreService initialize failed"; - return 1; - } - // QObject::connect(&app, &QCoreApplication::aboutToQuit, coreService, &CoreService::shutdown); - - // ======================================================================== - // 创建并注册 D-Bus 服务（仅在 CoreRouter 角色中启用） - // ======================================================================== - if (role == CoreService::Role::CoreRouter) { - const QString serviceName = QStringLiteral("com.softbus.Daemon"); - const QString objectPath = QStringLiteral("/com/softbus/Daemon"); - - IpcDBus* ipc = new IpcDBus(coreService, roleString, &app); - - auto pickConnection = [&]() -> QDBusConnection { - if (dbusBus == QStringLiteral("system")) { - return QDBusConnection::systemBus(); - } - if (dbusBus == QStringLiteral("session")) { - return QDBusConnection::sessionBus(); - } - // auto: 优先 system（更符合 daemon），失败则退回 session（方便本地开发） - QDBusConnection conn = QDBusConnection::systemBus(); - if (conn.isConnected()) { - return conn; - } - return QDBusConnection::sessionBus(); - }; - - QDBusConnection connection = pickConnection(); - if (!connection.isConnected()) { - LOG_ERROR() << "Cannot connect to D-Bus. system error=" - << QDBusConnection::systemBus().lastError().message() - << " session error=" << QDBusConnection::sessionBus().lastError().message(); - return 1; - } - - if (!connection.registerService(serviceName)) { - LOG_ERROR() << "Failed to register D-Bus service:" << serviceName - << "error=" << connection.lastError().message(); - return 1; - } - - if (!connection.registerObject(objectPath, - ipc, - QDBusConnection::ExportAllSlots | QDBusConnection::ExportAllSignals)) { - LOG_ERROR() << "Failed to register D-Bus object:" << objectPath - << "error=" << connection.lastError().message(); - return 1; - } - - LOG_INFO() << "D-Bus ready. bus=" - << (connection.name() == QDBusConnection::systemBus().name() ? "system" : "session") - << " service=" << serviceName - << " object=" << objectPath; - } - - - - // ================================== - // 处理系统信号，确保在接收到 SIGINT 或 SIGTERM 时能够优雅地退出 - // ================================== - - signal(SIGINT, signalHandler); - signal(SIGTERM, signalHandler); - - int ret = app.exec(); - return ret; -} +#include +#include +#include +#include +#include +#include "src/utils/logs/logging.h" +#include "src/core/CoreService.h" +#include "src/api/ipc_dbus/IpcDBus.h" +#include "src/api/http/RestApiServer.h" +#include "src/device_bus/DeviceBus.h" + +void signalHandler(int signal) +{ + LOG_INFO() << "Received signal:" << signal; + CoreService* coreService = CoreService::instance(); + if (coreService) { + coreService->shutdown(); + } + + QCoreApplication::quit(); +} + +int main(int argc, char *argv[]) +{ + // ================================== + // 创建并配置 Qt 应用程序对象 + // ================================== + QCoreApplication app(argc, argv); + + app.setApplicationName("soft_bus_daemon"); + app.setApplicationVersion("1.1.0"); + app.setOrganizationName("soft_bus"); + + // ================================== + // 设置 Qt 应用程序信息 + // ================================== + QCommandLineParser parser; + parser.setApplicationDescription(QStringLiteral("Soft Bus Daemon")); + parser.addHelpOption(); + parser.addVersionOption(); + QCommandLineOption roleOption(QStringLiteral("role"), + QStringLiteral("Daemon role (CoreRouter, CollectorSerial, CollectorCan, Collector, SqlServer)"), + QStringLiteral("role"), + QStringLiteral("CoreRouter")); + parser.addOption(roleOption); + QCommandLineOption dbusBusOption(QStringLiteral("dbus-bus"), + QStringLiteral("DBus bus to use: system|session|auto (default: auto)"), + QStringLiteral("bus"), + QStringLiteral("auto")); + parser.addOption(dbusBusOption); + parser.process(app); + + + const QString roleString = parser.value(roleOption).trimmed(); + const QString dbusBus = parser.value(dbusBusOption).trimmed(); + CoreService::Role role = CoreService::Role::CoreRouter; // 默认角色 + if (roleString == QStringLiteral("CoreRouter")) { + role = CoreService::Role::CoreRouter; + } else if (roleString == QStringLiteral("CollectorSerial")) { + role = CoreService::Role::CollectorSerial; + } else if (roleString == QStringLiteral("CollectorCan")) { + role = CoreService::Role::CollectorCan; + } else if (roleString == QStringLiteral("Collector")) { + role = CoreService::Role::Collector; + } else if (roleString == QStringLiteral("SqlServer")) { + role = CoreService::Role::SqlServer; + } else { + LOG_ERROR() << "Invalid role:" << roleString + << "valid roles are: CoreRouter, CollectorSerial, CollectorCan, Collector, SqlServer"; + return 1; + } + // 开启coreserive的初始化工作|线程 + // QThread* coreServiceThread = new QThread(); + // CoreService* coreService = new CoreService(); + // coreService->moveToThread(coreServiceThread); + // QObject::connect(coreServiceThread, &QThread::started, coreService, &CoreService::initialize); + // QObject::connect(coreServiceThread, &QThread::finished, coreService, &CoreService::deleteLater); + // QObject::connect(coreServiceThread, &QThread::finished, coreServiceThread, &QThread::deleteLater); + // coreServiceThread->start(); + + CoreService* coreService = CoreService::instance(); + if (!coreService->initialize()) { + LOG_ERROR() << "CoreService initialize failed"; + return 1; + } + // QObject::connect(&app, &QCoreApplication::aboutToQuit, coreService, &CoreService::shutdown); + + // ======================================================================== + // 创建并注册 D-Bus 服务（仅在 CoreRouter 角色中启用） + // ======================================================================== + if (role == CoreService::Role::CoreRouter) { + const QString serviceName = QStringLiteral("com.softbus.Daemon"); + const QString objectPath = QStringLiteral("/com/softbus/Daemon"); + + IpcDBus* ipc = new IpcDBus(coreService, roleString, &app); + + auto pickConnection = [&]() -> QDBusConnection { + if (dbusBus == QStringLiteral("system")) { + return QDBusConnection::systemBus(); + } + if (dbusBus == QStringLiteral("session")) { + return QDBusConnection::sessionBus(); + } + // auto: 优先 system（更符合 daemon），失败则退回 session（方便本地开发） + QDBusConnection conn = QDBusConnection::systemBus(); + if (conn.isConnected()) { + return conn; + } + return QDBusConnection::sessionBus(); + }; + + QDBusConnection connection = pickConnection(); + if (!connection.isConnected()) { + LOG_ERROR() << "Cannot connect to D-Bus. system error=" + << QDBusConnection::systemBus().lastError().message() + << " session error=" << QDBusConnection::sessionBus().lastError().message(); + return 1; + } + + if (!connection.registerService(serviceName)) { + LOG_ERROR() << "Failed to register D-Bus service:" << serviceName + << "error=" << connection.lastError().message(); + return 1; + } + + if (!connection.registerObject(objectPath, + ipc, + QDBusConnection::ExportAllSlots | QDBusConnection::ExportAllSignals)) { + LOG_ERROR() << "Failed to register D-Bus object:" << objectPath + << "error=" << connection.lastError().message(); + return 1; + } + + LOG_INFO() << "D-Bus ready. bus=" + << (connection.name() == QDBusConnection::systemBus().name() ? "system" : "session") + << " service=" << serviceName + << " object=" << objectPath; + + auto* restApi = new softbus::api::http::RestApiServer(&app); + QString restErr; + if (!restApi->tryStartFromConfigFile(softbus::device_bus::DeviceBus::instance().configPath(), &restErr)) { + LOG_ERROR() << "RestApiServer: start failed:" << restErr; + return 1; + } + QObject::connect(&app, &QCoreApplication::aboutToQuit, restApi, [restApi]() { restApi->stop(); }); + } + + + + // ================================== + // 处理系统信号，确保在接收到 SIGINT 或 SIGTERM 时能够优雅地退出 + // ================================== + + signal(SIGINT, signalHandler); + signal(SIGTERM, signalHandler); + + int ret = app.exec(); + return ret; +} diff --git a/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp index 4961d7e..1a845c8 100644 --- a/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp +++ b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.cpp @@ -2,7 +2,6 @@ #include -#include "message_bus/pipeline/PipelineContext.h" #include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" namespace softbus::message_bus::pipeline::protocol @@ -11,7 +10,7 @@ namespace softbus::message_bus::pipeline::protocol namespace { -using softbus::message_bus::pipeline::BusDirection; +using softbus::core::models::BusDirection; using softbus::message_bus::pipeline::protocol::modbusrtu::crc16; using softbus::message_bus::pipeline::protocol::modbusrtu::crcOk; @@ -325,103 +324,7 @@ bool encodeModbusRtu(const ModbusRtuPdu& pdu, QByteArray& out, QString* errMsg) } return true; } - if (const auto* req = std::get_if(&pdu)) { - out.append(static_cast(req->unitId)); - out.append(static_cast(0x05)); - out.append(static_cast((req->address >> 8) & 0xFF)); - out.append(static_cast(req->address & 0xFF)); - const std::uint16_t coilWord = req->value ? 0xFF00 : 0x0000; - out.append(static_cast((coilWord >> 8) & 0xFF)); - out.append(static_cast(coilWord & 0xFF)); - appendCrcLe(out); - return true; - } - if (const auto* resp = std::get_if(&pdu)) { - out.append(static_cast(resp->unitId)); - out.append(static_cast(0x05)); - out.append(static_cast((resp->address >> 8) & 0xFF)); - out.append(static_cast(resp->address & 0xFF)); - const std::uint16_t coilWord = resp->value ? 0xFF00 : 0x0000; - out.append(static_cast((coilWord >> 8) & 0xFF)); - out.append(static_cast(coilWord & 0xFF)); - appendCrcLe(out); - return true; - } - if (const auto* req = std::get_if(&pdu)) { - out.append(static_cast(req->unitId)); - out.append(static_cast(0x06)); - out.append(static_cast((req->address >> 8) & 0xFF)); - out.append(static_cast(req->address & 0xFF)); - out.append(static_cast((req->value >> 8) & 0xFF)); - out.append(static_cast(req->value & 0xFF)); - appendCrcLe(out); - return true; - } - if (const auto* resp = std::get_if(&pdu)) { - out.append(static_cast(resp->unitId)); - out.append(static_cast(0x06)); - out.append(static_cast((resp->address >> 8) & 0xFF)); - out.append(static_cast(resp->address & 0xFF)); - out.append(static_cast((resp->value >> 8) & 0xFF)); - out.append(static_cast(resp->value & 0xFF)); - appendCrcLe(out); - return true; - } - if (const auto* req = std::get_if(&pdu)) { - if (req->values.size() > 255) { - return fail("too_many_bytes"); - } - out.append(static_cast(req->unitId)); - out.append(static_cast(0x0F)); - out.append(static_cast((req->startAddress >> 8) & 0xFF)); - out.append(static_cast(req->startAddress & 0xFF)); - out.append(static_cast((req->quantity >> 8) & 0xFF)); - out.append(static_cast(req->quantity & 0xFF)); - out.append(static_cast(req->values.size())); - for (std::uint8_t b : req->values) { - out.append(static_cast(b)); - } - appendCrcLe(out); - return true; - } - if (const auto* resp = std::get_if(&pdu)) { - out.append(static_cast(resp->unitId)); - out.append(static_cast(0x0F)); - out.append(static_cast((resp->startAddress >> 8) & 0xFF)); - out.append(static_cast(resp->startAddress & 0xFF)); - out.append(static_cast((resp->quantity >> 8) & 0xFF)); - out.append(static_cast(resp->quantity & 0xFF)); - appendCrcLe(out); - return true; - } - if (const auto* req = std::get_if(&pdu)) { - if (req->values.size() > 127) { - return fail("too_many_registers"); - } - out.append(static_cast(req->unitId)); - out.append(static_cast(0x10)); - out.append(static_cast((req->startAddress >> 8) & 0xFF)); - out.append(static_cast(req->startAddress & 0xFF)); - out.append(static_cast((req->quantity >> 8) & 0xFF)); - out.append(static_cast(req->quantity & 0xFF)); - out.append(static_cast(req->values.size() * 2)); - for (std::uint16_t v : req->values) { - out.append(static_cast((v >> 8) & 0xFF)); - out.append(static_cast(v & 0xFF)); - } - appendCrcLe(out); - return true; - } - if (const auto* resp = std::get_if(&pdu)) { - out.append(static_cast(resp->unitId)); - out.append(static_cast(0x10)); - out.append(static_cast((resp->startAddress >> 8) & 0xFF)); - out.append(static_cast(resp->startAddress & 0xFF)); - out.append(static_cast((resp->quantity >> 8) & 0xFF)); - out.append(static_cast(resp->quantity & 0xFF)); - appendCrcLe(out); - return true; - } + // TODO: re-enable FC05/06/0F/10 encode branches after type-name collision investigation. return fail("unsupported_encode"); } diff --git a/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h index 0be068c..b70cce3 100644 --- a/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h +++ b/plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h @@ -6,17 +6,13 @@ #include #include +#include "core/models/RawBusMessage.h" #include "plugins/protocols/modbus_rtu/types/ModbusRtuPdu.h" -namespace softbus::message_bus::pipeline -{ -enum class BusDirection; -} - namespace softbus::message_bus::pipeline::protocol { -bool decodeModbusRtu(softbus::message_bus::pipeline::BusDirection dir, +bool decodeModbusRtu(softbus::core::models::BusDirection dir, const std::uint8_t* p, std::size_t n, ModbusRtuPdu& out, diff --git a/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp index 658d765..f2dab6c 100644 --- a/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp +++ b/plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.cpp @@ -1,7 +1,7 @@ #include "plugins/protocols/modbus_rtu/parser/ModbusRtuProtocolPlugin.h" #include "core/plugin_system/IProtocolPlugin.h" -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/RawBusMessage.h" #include "message_bus/pipeline/protocol/ProtocolEnvelope.h" #include "plugins/protocols/modbus_rtu/codec/ModbusRtuCodec.h" #include "plugins/protocols/modbus_rtu/framer/ModbusRtuFraming.h" @@ -49,11 +49,10 @@ overloaded(Ts...) -> overloaded; class ModbusRtuSession final : public IProtocolSession { public: - bool parse(const softbus::message_bus::pipeline::PipelineContext& ctx, + bool parse(const softbus::core::models::RawBusMessage& ctx, softbus::message_bus::pipeline::protocol::ProtocolEnvelope& out) override { - using softbus::message_bus::pipeline::FrameKind; - if (ctx.frameKind != FrameKind::CompleteFrame || !ctx.payload.valid()) { + if (!ctx.payload.valid()) { return false; } const std::size_t n = ctx.payloadSize ? ctx.payloadSize : ctx.payload.length; @@ -63,7 +62,11 @@ public: } ModbusRtuPdu pdu; - if (!decodeModbusRtu(ctx.direction, p, n, pdu, nullptr)) { + if (!decodeModbusRtu(ctx.direction, + p, + n, + pdu, + nullptr)) { out.family = ProtocolFamily::None; out.pdu = {}; return false; diff --git a/src/api/http/RestApiCommandHttp.cpp b/src/api/http/RestApiCommandHttp.cpp new file mode 100644 index 0000000..a74bbae --- /dev/null +++ b/src/api/http/RestApiCommandHttp.cpp @@ -0,0 +1,43 @@ +#include "api/http/RestApiCommandHttp.h" + +#include +#include +#include +#include + +#include "api/ipc_dbus/CommandDispatcher.h" + +namespace softbus::api::http +{ + +QHttpServerResponse makeCommandHttpResponse(const QString& id, const softbus::api::ipc::DispatchResult& r) +{ + QJsonObject o; + o.insert(QStringLiteral("id"), id); + o.insert(QStringLiteral("code"), r.code); + o.insert(QStringLiteral("message"), r.message); + o.insert(QStringLiteral("data"), r.data); + const auto httpStatus = + r.code == 0 ? QHttpServerResponse::StatusCode::Ok : QHttpServerResponse::StatusCode::BadRequest; + return QHttpServerResponse(o, httpStatus); +} + +QHttpServerResponse handleExecuteCommandHttpBody(const QByteArray& body, softbus::api::ipc::CommandDispatcher& dispatcher) +{ + QJsonParseError err; + const QJsonDocument doc = QJsonDocument::fromJson(body, &err); + if (err.error != QJsonParseError::NoError || !doc.isObject()) { + return makeCommandHttpResponse(QString(), {4000, QStringLiteral("invalid_json"), {}}); + } + const QJsonObject req = doc.object(); + const QString id = req.value(QStringLiteral("id")).toString(); + const QString action = req.value(QStringLiteral("action")).toString(); + const QJsonObject params = req.value(QStringLiteral("params")).toObject(); + if (action.isEmpty()) { + return makeCommandHttpResponse(id, {4003, QStringLiteral("invalid_action"), {}}); + } + const auto result = dispatcher.dispatch(action, params); + return makeCommandHttpResponse(id, result); +} + +} // namespace softbus::api::http diff --git a/src/api/http/RestApiCommandHttp.h b/src/api/http/RestApiCommandHttp.h new file mode 100644 index 0000000..8e87bf2 --- /dev/null +++ b/src/api/http/RestApiCommandHttp.h @@ -0,0 +1,22 @@ +#pragma once + +#include +#include +#include + +namespace softbus::api::ipc +{ +struct DispatchResult; +class CommandDispatcher; +} // namespace softbus::api::ipc + +namespace softbus::api::http +{ + +/// Same envelope as D-Bus `executeCommand` response body (id/code/message/data). +QHttpServerResponse makeCommandHttpResponse(const QString& id, const softbus::api::ipc::DispatchResult& r); + +/// POST body same shape as D-Bus request: { "id", "action", "params" }. +QHttpServerResponse handleExecuteCommandHttpBody(const QByteArray& body, softbus::api::ipc::CommandDispatcher& dispatcher); + +} // namespace softbus::api::http diff --git a/src/api/http/RestApiHttpConfig.cpp b/src/api/http/RestApiHttpConfig.cpp new file mode 100644 index 0000000..1f5b008 --- /dev/null +++ b/src/api/http/RestApiHttpConfig.cpp @@ -0,0 +1,67 @@ +#include "api/http/RestApiHttpConfig.h" + +#include +#include +#include +#include +#include + +namespace softbus::api::http +{ + +bool loadRestApiHttpListenOptions(const QString& configPath, RestApiHttpListenOptions* out, QString* err) +{ + if (!out) { + if (err) { + *err = QStringLiteral("null_output"); + } + return false; + } + *out = {}; + + if (configPath.isEmpty()) { + if (err) { + *err = QStringLiteral("config_path_empty"); + } + return false; + } + + QFile f(configPath); + if (!f.exists()) { + return true; + } + if (!f.open(QIODevice::ReadOnly | QIODevice::Text)) { + if (err) { + *err = QStringLiteral("config_open_failed"); + } + return false; + } + const QByteArray raw = f.readAll(); + f.close(); + + QJsonParseError pe; + const QJsonDocument doc = QJsonDocument::fromJson(raw, &pe); + if (pe.error != QJsonParseError::NoError || !doc.isObject()) { + if (err) { + *err = QStringLiteral("config_json_invalid"); + } + return false; + } + const QJsonObject root = doc.object(); + const QJsonObject api = root.value(QStringLiteral("api")).toObject(); + const QJsonObject httpCfg = api.value(QStringLiteral("http")).toObject(); + if (httpCfg.isEmpty()) { + return true; + } + + out->sectionPresent = true; + out->enabled = httpCfg.value(QStringLiteral("enabled")).toBool(false); + out->bindHost = httpCfg.value(QStringLiteral("bind")).toString(QStringLiteral("127.0.0.1")); + if (out->bindHost.isEmpty()) { + out->bindHost = QStringLiteral("127.0.0.1"); + } + out->port = httpCfg.value(QStringLiteral("port")).toInt(18080); + return true; +} + +} // namespace softbus::api::http diff --git a/src/api/http/RestApiHttpConfig.h b/src/api/http/RestApiHttpConfig.h new file mode 100644 index 0000000..b5c90a4 --- /dev/null +++ b/src/api/http/RestApiHttpConfig.h @@ -0,0 +1,21 @@ +#pragma once + +#include + +namespace softbus::api::http +{ + +/// Parsed `api.http` block from daemon_config.json. +struct RestApiHttpListenOptions +{ + bool sectionPresent{false}; + bool enabled{false}; + QString bindHost; + int port{18080}; +}; + +/// Reads and validates `api.http`. On success returns true; on missing file returns true with sectionPresent=false. +/// Hard errors (open/parse) set err and return false. +bool loadRestApiHttpListenOptions(const QString& configPath, RestApiHttpListenOptions* out, QString* err = nullptr); + +} // namespace softbus::api::http diff --git a/src/api/http/RestApiRoutes.cpp b/src/api/http/RestApiRoutes.cpp new file mode 100644 index 0000000..1511e90 --- /dev/null +++ b/src/api/http/RestApiRoutes.cpp @@ -0,0 +1,44 @@ +#include "api/http/RestApiRoutes.h" + +#include +#include +#include +#include +#include +#include + +#include "api/http/RestApiCommandHttp.h" +#include "api/ipc_dbus/CommandDispatcher.h" + +namespace softbus::api::http +{ + +void registerSoftbusRestApiRoutes(QHttpServer* server, + QObject* routeContext, + softbus::api::ipc::CommandDispatcher* dispatcher) +{ + if (!server || !routeContext || !dispatcher) { + return; + } + + server->route(QStringLiteral("/health"), QHttpServerRequest::Method::Get, routeContext, []() { + return QHttpServerResponse(QJsonObject{{QStringLiteral("ok"), true}}, QHttpServerResponse::StatusCode::Ok); + }); + + server->route(QStringLiteral("/v1/status"), QHttpServerRequest::Method::Get, routeContext, [dispatcher]() { + return makeCommandHttpResponse(QString(), dispatcher->dispatch(QStringLiteral("get_status"), {})); + }); + + server->route(QStringLiteral("/v1/command"), QHttpServerRequest::Method::Post, routeContext, + [dispatcher](const QHttpServerRequest& request) { + return handleExecuteCommandHttpBody(request.body(), *dispatcher); + }); + + server->setMissingHandler(routeContext, [](const QHttpServerRequest&, QHttpServerResponder& responder) { + QHttpServerResponse resp(QJsonObject{{QStringLiteral("error"), QStringLiteral("not_found")}}, + QHttpServerResponse::StatusCode::NotFound); + responder.sendResponse(std::move(resp)); + }); +} + +} // namespace softbus::api::http diff --git a/src/api/http/RestApiRoutes.h b/src/api/http/RestApiRoutes.h new file mode 100644 index 0000000..6f8900e --- /dev/null +++ b/src/api/http/RestApiRoutes.h @@ -0,0 +1,19 @@ +#pragma once + +class QObject; +class QHttpServer; + +namespace softbus::api::ipc +{ +class CommandDispatcher; +} // namespace softbus::api::ipc + +namespace softbus::api::http +{ + +/// Registers REST routes on `server`. `routeContext` is the QObject passed to QHttpServer::route (lifetime). +void registerSoftbusRestApiRoutes(QHttpServer* server, + QObject* routeContext, + softbus::api::ipc::CommandDispatcher* dispatcher); + +} // namespace softbus::api::http diff --git a/src/api/http/RestApiServer.cpp b/src/api/http/RestApiServer.cpp new file mode 100644 index 0000000..920bfcf --- /dev/null +++ b/src/api/http/RestApiServer.cpp @@ -0,0 +1,115 @@ +#include "api/http/RestApiServer.h" + +#include +#include +#include + +#include "api/http/RestApiHttpConfig.h" +#include "api/http/RestApiRoutes.h" +#include "api/ipc_dbus/CommandDispatcher.h" +#include "utils/logs/logging.h" + +namespace softbus::api::http +{ + +struct RestApiServer::Impl +{ + std::unique_ptr dispatcher; + std::unique_ptr http; + std::unique_ptr tcp; +}; + +RestApiServer::RestApiServer(QObject* parent) + : QObject(parent) + , m_impl(std::make_unique()) +{ + m_impl->dispatcher = std::make_unique(); +} + +RestApiServer::~RestApiServer() +{ + stop(); +} + +void RestApiServer::stop() +{ + if (!m_impl || !m_impl->http) { + return; + } + m_impl->http->disconnect(); + m_impl->http.reset(); + m_impl->tcp.reset(); +} + +bool RestApiServer::tryStartFromConfigFile(const QString& configPath, QString* err) +{ + if (!m_impl || !m_impl->dispatcher) { + if (err) { + *err = QStringLiteral("internal_error"); + } + return false; + } + + RestApiHttpListenOptions opts; + if (!loadRestApiHttpListenOptions(configPath, &opts, err)) { + return false; + } + if (!opts.sectionPresent) { + LOG_INFO() << "RestApiServer: no api.http section, HTTP disabled"; + return true; + } + if (!opts.enabled) { + LOG_INFO() << "RestApiServer: api.http.enabled is false or missing, HTTP disabled"; + return true; + } + + if (opts.port <= 0 || opts.port > 65535) { + if (err) { + *err = QStringLiteral("invalid_port"); + } + return false; + } + + stop(); + + m_impl->http = std::make_unique(this); + m_impl->tcp = std::make_unique(); + + QHttpServer* srv = m_impl->http.get(); + registerSoftbusRestApiRoutes(srv, this, m_impl->dispatcher.get()); + + const QHostAddress addr(opts.bindHost); + if (addr.isNull()) { + if (err) { + *err = QStringLiteral("invalid_bind_address"); + } + stop(); + return false; + } + + if (!m_impl->tcp->listen(addr, static_cast(opts.port))) { + if (err) { + *err = QStringLiteral("tcp_listen_failed"); + } + LOG_ERROR() << "RestApiServer: listen failed bind=" << opts.bindHost << "port=" << opts.port + << "err=" << m_impl->tcp->errorString(); + stop(); + return false; + } + + if (!srv->bind(m_impl->tcp.get())) { + if (err) { + *err = QStringLiteral("http_bind_failed"); + } + LOG_ERROR() << "RestApiServer: QHttpServer::bind failed"; + stop(); + return false; + } + + (void)m_impl->tcp.release(); + + LOG_INFO() << "RestApiServer: listening http://" << opts.bindHost << ":" << opts.port; + return true; +} + +} // namespace softbus::api::http diff --git a/src/api/http/RestApiServer.h b/src/api/http/RestApiServer.h new file mode 100644 index 0000000..29eeff3 --- /dev/null +++ b/src/api/http/RestApiServer.h @@ -0,0 +1,29 @@ +#pragma once + +#include +#include +#include + +namespace softbus::api::http +{ + +class RestApiServer final : public QObject +{ + Q_OBJECT + +public: + explicit RestApiServer(QObject* parent = nullptr); + ~RestApiServer() override; + + /// Reads `api.http` from JSON config. If disabled or missing, returns true and does nothing. + /// If enabled but listen/bind fails, returns false and sets err. + bool tryStartFromConfigFile(const QString& configPath, QString* err = nullptr); + + void stop(); + +private: + struct Impl; + std::unique_ptr m_impl; +}; + +} // namespace softbus::api::http diff --git a/src/core/models/RawBusMessage.h b/src/core/models/RawBusMessage.h new file mode 100644 index 0000000..3b38e70 --- /dev/null +++ b/src/core/models/RawBusMessage.h @@ -0,0 +1,38 @@ +#pragma once + +#include + +#include +#include +#include + +#include "message_bus/pipeline/PayloadRef.h" + +namespace softbus::core::models +{ + +enum class BusDirection +{ + Unknown, + Upstream, + Downstream, +}; + +struct RawBusMessage +{ + int deviceId{-1}; + QString endpoint; + QString stableKey; + QDateTime timestamp{QDateTime::currentDateTimeUtc()}; + QString protocolHint; + QString traceId; + + softbus::message_bus::pipeline::PayloadRef payload; + std::size_t payloadSize{0}; + + BusDirection direction{BusDirection::Unknown}; +}; + +} // namespace softbus::core::models + +Q_DECLARE_METATYPE(softbus::core::models::RawBusMessage) diff --git a/src/core/plugin_system/IProtocolPlugin.h b/src/core/plugin_system/IProtocolPlugin.h index 2107e36..020bd10 100644 --- a/src/core/plugin_system/IProtocolPlugin.h +++ b/src/core/plugin_system/IProtocolPlugin.h @@ -4,8 +4,8 @@ #include +#include "core/models/RawBusMessage.h" #include "message_bus/pipeline/protocol/ProtocolEnvelope.h" -#include "message_bus/pipeline/PipelineContext.h" namespace softbus::core::plugin_system { @@ -14,7 +14,7 @@ class IProtocolSession { public: virtual ~IProtocolSession() = default; - virtual bool parse(const softbus::message_bus::pipeline::PipelineContext& ctx, + virtual bool parse(const softbus::core::models::RawBusMessage& ctx, softbus::message_bus::pipeline::protocol::ProtocolEnvelope& out) = 0; virtual void reset() = 0; }; diff --git a/src/devices/physical/SerialDevice.cpp b/src/devices/physical/SerialDevice.cpp index 7993672..5ef6f64 100644 --- a/src/devices/physical/SerialDevice.cpp +++ b/src/devices/physical/SerialDevice.cpp @@ -7,9 +7,9 @@ #include #include "core/memory/MemoryPool.h" +#include "core/models/RawBusMessage.h" #include "hardware/drivers/serial/SerialQtDriver.h" #include "message_bus/ingress/IIngressPort.h" -#include "message_bus/pipeline/PipelineContext.h" #include "utils/logs/logging.h" using softbus::hardware::drivers::serial::SerialQtDriver; @@ -113,11 +113,10 @@ void SerialDevice::onRead(const std::vector& data) // @test code LOG_INFO() << "SerialDevice: onRead data size" << data.size() << "endpoint" << m_config.endpoint; // 将数据推送到 ingress 队列 包含端口名、协议、数据、时间戳、traceId - softbus::message_bus::pipeline::PipelineContext ctx; - ctx.direction = softbus::message_bus::pipeline::BusDirection::Upstream; + softbus::core::models::RawBusMessage ctx; + ctx.direction = softbus::core::models::BusDirection::Upstream; ctx.endpoint = m_config.endpoint; ctx.protocolHint = m_info.protocol; - ctx.frameKind = softbus::message_bus::pipeline::FrameKind::StreamChunk; ctx.payload.block = std::move(block); ctx.payload.length = data.size(); ctx.payloadSize = data.size(); diff --git a/src/message_bus/ingress/DriverIngressAdapter.cpp b/src/message_bus/ingress/DriverIngressAdapter.cpp index af5422e..cc12aff 100644 --- a/src/message_bus/ingress/DriverIngressAdapter.cpp +++ b/src/message_bus/ingress/DriverIngressAdapter.cpp @@ -12,12 +12,14 @@ DriverIngressAdapter::DriverIngressAdapter(std::shared_ptrenqueueIngress(std::move(ctx)); + softbus::message_bus::pipeline::PipelineContext pipelineCtx; + pipelineCtx.raw() = std::move(ctx); + return m_engine->enqueueIngress(std::move(pipelineCtx)); } void DriverIngressAdapter::reportError(const QString& endpoint, const QString& error) diff --git a/src/message_bus/ingress/DriverIngressAdapter.h b/src/message_bus/ingress/DriverIngressAdapter.h index 65a5e25..8ce35dc 100644 --- a/src/message_bus/ingress/DriverIngressAdapter.h +++ b/src/message_bus/ingress/DriverIngressAdapter.h @@ -16,7 +16,7 @@ public: DriverIngressAdapter(std::shared_ptr pool, softbus::message_bus::pipeline::PipelineEngine* engine); - bool push(softbus::message_bus::pipeline::PipelineContext ctx) override; + bool push(softbus::core::models::RawBusMessage ctx) override; void reportError(const QString& endpoint, const QString& error) override; softbus::core::memory::MemoryPool& pool() const { return *m_pool; } diff --git a/src/message_bus/ingress/IIngressPort.h b/src/message_bus/ingress/IIngressPort.h index 5eb5710..36f9b10 100644 --- a/src/message_bus/ingress/IIngressPort.h +++ b/src/message_bus/ingress/IIngressPort.h @@ -2,7 +2,7 @@ #include -#include "message_bus/pipeline/PipelineContext.h" +#include "core/models/RawBusMessage.h" namespace softbus::message_bus::ingress { @@ -12,7 +12,7 @@ class IIngressPort public: virtual ~IIngressPort() = default; - virtual bool push(softbus::message_bus::pipeline::PipelineContext ctx) = 0; + virtual bool push(softbus::core::models::RawBusMessage ctx) = 0; virtual void reportError(const QString& endpoint, const QString& error) = 0; }; diff --git a/src/message_bus/pipeline/PipelineContext.h b/src/message_bus/pipeline/PipelineContext.h index 1a09e44..0024ae8 100644 --- a/src/message_bus/pipeline/PipelineContext.h +++ b/src/message_bus/pipeline/PipelineContext.h @@ -1,14 +1,9 @@ #pragma once -// 定义管道上下文 包含设备ID、端点、时间戳、协议提示、跟踪ID、有效载荷、有效载荷大小 -#include #include -#include #include -#include - -#include "message_bus/pipeline/PayloadRef.h" +#include "core/models/RawBusMessage.h" namespace softbus::message_bus::pipeline { @@ -19,29 +14,14 @@ enum class FrameKind CompleteFrame, }; -enum class BusDirection +using BusDirection = softbus::core::models::BusDirection; + +struct PipelineContext : public softbus::core::models::RawBusMessage { - Unknown, - Upstream, - Downstream, -}; - -struct PipelineContext -{ - int deviceId{-1}; // 设备ID - QString stableKey; - QString endpoint; // 端点 - QDateTime timestamp{QDateTime::currentDateTimeUtc()}; // 时间戳 - QString protocolHint; // 协议提示 - QString traceId; // 跟踪ID - - PayloadRef payload; // 有效载荷 - std::size_t payloadSize{0}; // 有效载荷大小 - FrameKind frameKind{FrameKind::StreamChunk}; - BusDirection direction{BusDirection::Unknown}; std::uint64_t frameSeq{0}; - + softbus::core::models::RawBusMessage& raw() { return *this; } + const softbus::core::models::RawBusMessage& raw() const { return *this; } }; } // namespace softbus::message_bus::pipeline diff --git a/src/message_bus/pipeline/PipelineEngine.cpp b/src/message_bus/pipeline/PipelineEngine.cpp index 233b79a..717c207 100644 --- a/src/message_bus/pipeline/PipelineEngine.cpp +++ b/src/message_bus/pipeline/PipelineEngine.cpp @@ -113,9 +113,11 @@ void PipelineEngine::start() // 合并队列是用来存储合并的帧的 // 合并线程是用来处理合并的帧的 if (m_rtConfig.parallelPipeline) { - m_framedQ = std::make_unique>( - m_rtConfig.framedQueueMax); + m_framedQ = std::make_unique>(m_rtConfig.framedQueueMax); for (int i = 0; i < m_rtConfig.parseWorkerCount; ++i) { + // 创建解析线程 解析线程从帧队列中取出帧 进行处理 + // emplace_back 是用来创建解析线程的 参数是 lambda 表达式 + // [this]() { parseWorkerLoop(); } 是一个 lambda 表达式 用来创建解析线程 m_parseThreads.emplace_back([this]() { parseWorkerLoop(); }); } if (m_rtConfig.orderedEgress) { @@ -503,7 +505,7 @@ std::vector PipelineEngine::stage2Parser(cons parserSession = it.value(); } softbus::message_bus::pipeline::protocol::ProtocolEnvelope envelope; // 协议封装 - if (!parserSession->parse(ctx, envelope)) { + if (!parserSession->parse(ctx.raw(), envelope)) { return {}; }