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

devicesbus modbus rtu

Browse Source
This commit is contained in:
flower_linux
2026-04-02 11:18:17 +08:00
parent 3b7368a2ab
commit 3341bd0e6b
7 changed files with 90 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
CMakeLists.txt
View File

@@ -119,14 +119,26 @@ qt_add_executable(softbus_daemon
src/core/plugin_system/IProtocolPlugin.h
src/core/plugin_system/PluginManager.h
src/core/plugin_system/PluginManager.cpp
src/core/plugin_system/builtins/ModbusRtuProtocolPlugin.h
src/core/plugin_system/builtins/ModbusRtuProtocolPlugin.cpp
src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h
src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.cpp
src/core/models/Types.h
src/core/models/MetadataDef.h
src/core/models/DataMapping.h
src/core/models/DOMMessage.h
src/core/metadata/MetadataRegistry.h
src/core/metadata/MetadataRegistry.cpp
src/core/metadata/ProfileRegistry.h
src/core/metadata/ProfileRegistry.cpp
# message bus pipeline
src/message_bus/pipeline/PayloadRef.h
src/message_bus/pipeline/PipelineContext.h
src/message_bus/pipeline/PipelineEngine.h
src/message_bus/pipeline/PipelineEngine.cpp
src/message_bus/pipeline/protocol/ProtocolEnvelope.h
src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuPdu.h
src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.h
src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp
src/message_bus/pipeline/stages/1_framers/IFramer.h
src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h
src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp
@@ -136,6 +148,8 @@ qt_add_executable(softbus_daemon
src/message_bus/pipeline/stages/1_framers/PassthroughFramer.cpp
src/message_bus/pipeline/stages/2_parsers/IParserStage.h
src/message_bus/pipeline/stages/3_filters/IFilterStage.h
src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h
src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp
src/message_bus/pipeline/stages/4_validators/IValidator.h
# message bus ingress/egress
@@ -202,6 +216,35 @@ if(SOFTBUS_BUILD_TESTS)
Qt6::Core
Qt6::Test
)
qt_add_executable(softbus_protocol_parse_filter_test
tests/protocol_parse_filter_test.cpp
src/core/metadata/MetadataRegistry.cpp
src/core/metadata/ProfileRegistry.cpp
src/message_bus/pipeline/stages/3_filters/ProtocolParseFilter.cpp
)
qt_add_executable(softbus_modbus_rtu_pdu_codec_test
tests/modbus_rtu_pdu_codec_test.cpp
src/message_bus/pipeline/protocol/modbusrtu/ModbusRtuCodec.cpp
src/message_bus/pipeline/stages/1_framers/ModbusRtuFraming.cpp
)
target_include_directories(softbus_modbus_rtu_pdu_codec_test PRIVATE
${CMAKE_SOURCE_DIR}
${CMAKE_SOURCE_DIR}/src
)
target_link_libraries(softbus_modbus_rtu_pdu_codec_test PRIVATE
Qt6::Core
Qt6::Test
)
target_include_directories(softbus_protocol_parse_filter_test PRIVATE
${CMAKE_SOURCE_DIR}
${CMAKE_SOURCE_DIR}/src
)
target_link_libraries(softbus_protocol_parse_filter_test PRIVATE
Qt6::Core
Qt6::Test
)
endif()
install(TARGETS softbus_daemon

BIN
bin/softbus_daemon
View File

Binary file not shown.

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

@@ -10,7 +10,7 @@
#include "message_bus/ingress/DriverIngressAdapter.h"
#include "message_bus/pipeline/PipelineEngine.h"
#include "core/plugin_system/PluginManager.h"
#include "core/plugin_system/builtins/ModbusRtuProtocolPlugin.h"
#include "message_bus/pipeline/protocol/modbusrtu/ModbusRtuProtocolPlugin.h"
#include "device_bus/manager/SerialDeviceManager.h"
#include "devices/DeviceTypes.h"
#include "utils/logs/logging.h"

1
src/devices/physical/SerialDevice.cpp
View File

@@ -114,6 +114,7 @@ void SerialDevice::onRead(const std::vector<std::uint8_t>& data)
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;
ctx.endpoint = m_config.endpoint;
ctx.protocolHint = m_info.protocol;
ctx.frameKind = softbus::message_bus::pipeline::FrameKind::StreamChunk;

8
src/message_bus/pipeline/PipelineContext.h
View File

@@ -8,8 +8,11 @@
#include <QJsonObject>
#include <QMetaType>
#include <QString>
#include <QVector>
#include "core/models/DOMMessage.h"
#include "message_bus/pipeline/PayloadRef.h"
#include "message_bus/pipeline/protocol/ProtocolEnvelope.h"
namespace softbus::message_bus::pipeline
{
@@ -43,7 +46,12 @@ struct PipelineContext
BusDirection direction{BusDirection::Unknown};
std::uint64_t frameSeq{0};
/// 强类型协议解析结果（推荐下游使用）；多协议扩展时只增不改 variant。
softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol;
/// 兼容旧逻辑与日志；新代码请优先使用 `protocol`，勿依赖此处 JSON 字段名。
QJsonObject parsed;
QVector<softbus::core::models::DOMMessage> domMessages;
QString framerError; // 成帧/Pipeline 旁路诊断（可选）
};

46
src/message_bus/pipeline/PipelineEngine.cpp
View File

@@ -7,6 +7,7 @@
#include <functional>
#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h"
#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
#include "utils/logs/logging.h"
namespace softbus::message_bus::pipeline
@@ -360,16 +361,7 @@ void PipelineEngine::parseWorkerLoop()
}
continue;
}
std::ostringstream oss;
oss << std::hex << std::setfill('0');
for (std::size_t i = 0; i < ctx.payload.block->size; ++i) {
oss << std::setw(2) << static_cast<int>(ctx.payload.block->data[i]);
if (i + 1 < ctx.payload.block->size) {
oss << ' ';
}
}
const QString hexPreview = QString::fromStdString(oss.str());
LOG_INFO() << "the data is"<< hexPreview;
const bool ok = runStages234(ctx);
if (m_rtConfig.orderedEgress && m_mergeQ) {
@@ -449,9 +441,39 @@ bool PipelineEngine::runStages234(PipelineContext& ctx)
if (ctx.frameKind != FrameKind::CompleteFrame) {
return false;
}
std::ostringstream oss;
oss << std::hex << std::setfill('0');
for (std::size_t i = 0; i < ctx.payload.block->size; ++i) {
oss << std::setw(2) << static_cast<int>(ctx.payload.block->data[i]);
if (i + 1 < ctx.payload.block->size) {
oss << ' ';
}
}
const QString hexPreview = QString::fromStdString(oss.str());
LOG_INFO() << "the data is"<< hexPreview;
LOG_INFO() << "the protocol hint is"<< ctx.protocolHint;
const QString directionText = (ctx.direction == BusDirection::Upstream)
? QStringLiteral("Upstream")
: (ctx.direction == BusDirection::Downstream)
? QStringLiteral("Downstream")
: QStringLiteral("Unknown");
LOG_INFO() << "the direction is" << directionText;
if (!stage2Parser(ctx)) {
return false;
}
QJsonObject parsed = ctx.parsed;
LOG_INFO() << "the parsed is"<< parsed;
softbus::message_bus::pipeline::protocol::ProtocolEnvelope protocol = ctx.protocol;
if (const auto* modbusPdu = std::get_if<softbus::message_bus::pipeline::protocol::ModbusRtuPdu>(&protocol.pdu)) {
LOG_INFO() << "the protocol text is" << softbus::message_bus::pipeline::protocol::toText(*modbusPdu);
} else {
LOG_INFO() << "the protocol text is" << QStringLiteral("ProtocolEnvelopePdu::monostate");
}
m_count++;
LOG_INFO() << "the count of command is" << m_count;
if (!stage3Filter(ctx)) {
return false;
}
@@ -504,8 +526,8 @@ bool PipelineEngine::stage2Parser(PipelineContext& ctx)
bool PipelineEngine::stage3Filter(PipelineContext& ctx)
{
(void)ctx;
return true;
static softbus::message_bus::pipeline::stages::filters::ProtocolParseFilter s_filter;
return s_filter.apply(ctx);
}
} // namespace softbus::message_bus::pipeline

1
src/message_bus/pipeline/PipelineEngine.h
View File

@@ -122,6 +122,7 @@ private:
std::mutex m_mergeStateMutex;
QHash<QString, MergeState> m_mergeByEndpoint;
int m_count{0}; // 计数器
};
} // namespace softbus::message_bus::pipeline