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

devices modbus rtu 1.0

Browse Source
This commit is contained in:
flower_linux
2026-04-07 14:20:04 +08:00
parent 28b5082742
commit 51175ffd18
38 changed files with 1006 additions and 745 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,7 +6,8 @@
#include <chrono>
#include <functional>
#include "message_bus/pipeline/stages/1_framers/ModbusRtuFraming.h"
#include "plugins/protocols/modbusrtu/framer/ModbusRtuFraming.h"
#include "plugins/protocols/modbusrtu/framer/ModbusRtuFramerPlugin.h"
#include "message_bus/pipeline/stages/3_filters/ProtocolParseFilter.h"
#include "utils/logs/logging.h"
@@ -21,8 +22,7 @@ struct EndpointFramingState
std::uint64_t nextSeq{1};
QString endpoint;
std::shared_ptr<softbus::core::memory::MemoryPool> pool;
softbus::message_bus::pipeline::stages::framers::ModbusRtuFramerConfig modbusCfg;
std::shared_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> pass;
std::function<void(PipelineContext&, std::vector<PipelineContext>&)> feed;
std::function<void(PipelineContext)> pushFramed;
explicit EndpointFramingState(std::size_t cap)
@@ -36,11 +36,6 @@ namespace
void endpointFramerThread(std::shared_ptr<EndpointFramingState> st)
{
using softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer;
using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu;
auto modbus = std::make_unique<ModbusRtuFramer>(st->pool, st->modbusCfg, st->endpoint);
while (st->running.load(std::memory_order_acquire)) {
PipelineContext ch;
if (!st->chunkQ.pop(ch)) {
@@ -48,10 +43,8 @@ void endpointFramerThread(std::shared_ptr<EndpointFramingState> st)
continue;
}
std::vector<PipelineContext> out;
if (protocolHintIsModbusRtu(ch.protocolHint)) {
modbus->feed(ch, out);
} else if (st->pass) {
st->pass->feed(ch, out);
if (st->feed) {
st->feed(ch, out);
}
for (auto& x : out) {
x.frameSeq = st->nextSeq++;
@@ -98,9 +91,9 @@ void PipelineEngine::loadRuntimeConfigFromJson(const QJsonObject& obj)
std::max(64, obj.value(QStringLiteral("framedQueueMax")).toInt(4096)));
m_rtConfig.maxReorderDepth = static_cast<std::size_t>(
std::max(16, obj.value(QStringLiteral("maxReorderDepth")).toInt(256)));
m_rtConfig.modbus.maxAdu = static_cast<std::size_t>(
m_rtConfig.modbusMaxAdu = static_cast<std::size_t>(
std::max(8, obj.value(QStringLiteral("modbusRtuMaxFrameBytes")).toInt(256)));
m_rtConfig.modbus.interFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0);
m_rtConfig.modbusInterFrameTimeoutMs = obj.value(QStringLiteral("modbusRtuInterFrameTimeoutMs")).toInt(0);
}
void PipelineEngine::start()
@@ -115,12 +108,15 @@ void PipelineEngine::start()
return;
}
m_modbusFramer =
std::make_unique<softbus::message_bus::pipeline::stages::framers::ModbusRtuFramer>(m_pool,
m_rtConfig.modbus,
QString{});
m_passthroughFramer =
std::make_shared<softbus::message_bus::pipeline::stages::framers::PassthroughFramer>(m_pool);
if (m_pluginManager) {
softbus::message_bus::pipeline::protocol::modbusrtu::ModbusRtuFramerPluginConfig cfg;
cfg.maxAdu = m_rtConfig.modbusMaxAdu;
cfg.interFrameTimeoutMs = m_rtConfig.modbusInterFrameTimeoutMs;
m_pluginManager->registerProtocolFramerPlugin(
softbus::message_bus::pipeline::protocol::modbusrtu::makeModbusRtuFramerPlugin(m_pool, cfg));
}
// 如果启用并行管道 则创建帧队列和解析线程 如果启用有序出口 则创建合并队列和合并线程
// 并行管道是用来处理多个端点的数据流的
@@ -182,8 +178,11 @@ void PipelineEngine::stop()
m_framedQ.reset();
m_mergeQ.reset();
m_modbusFramer.reset();
m_passthroughFramer.reset();
{
std::lock_guard<std::mutex> lk(m_framerSessionMutex);
m_framerSessions.clear();
}
}
// 功能 ：清空 ingress 队列
// 参数 ：无
@@ -266,8 +265,9 @@ void PipelineEngine::ensureEndpointFramer(const QString& endpoint)
auto st = std::make_shared<EndpointFramingState>(qcap > 0 ? qcap : 4096);
st->endpoint = endpoint;
st->pool = m_pool;
st->modbusCfg = m_rtConfig.modbus;
st->pass = m_passthroughFramer;
st->feed = [this](PipelineContext& in, std::vector<PipelineContext>& out) {
this->dispatchFramer(in, out);
};
st->pushFramed = [this](PipelineContext c) {
if (!m_framedQ) {
return;
@@ -286,17 +286,30 @@ void PipelineEngine::ensureEndpointFramer(const QString& endpoint)
// 逻辑 ：根据协议提示 选择合适的帧解析器 进行帧解析
void PipelineEngine::dispatchFramer(PipelineContext& chunkIn, std::vector<PipelineContext>& completeFramesOut)
{
using softbus::message_bus::pipeline::stages::framers::protocolHintIsModbusRtu;
// 根据协议提示 选择合适的帧解析器 进行帧解析
if (protocolHintIsModbusRtu(chunkIn.protocolHint)) {
if (m_modbusFramer) {
m_modbusFramer->feed(chunkIn, completeFramesOut);
if (m_pluginManager) {
auto plugin = m_pluginManager->selectProtocolFramerPlugin(chunkIn.protocolHint);
if (plugin) {
plugin->bindMemoryPool(m_pool);
const QString key = chunkIn.endpoint + QStringLiteral("|") + plugin->pluginId();
std::lock_guard<std::mutex> lk(m_framerSessionMutex);
auto it = m_framerSessions.find(key);
if (it == m_framerSessions.end() || !it.value()) {
QJsonObject params;
params.insert(QStringLiteral("modbusRtuMaxFrameBytes"), static_cast<int>(m_rtConfig.modbusMaxAdu));
params.insert(QStringLiteral("modbusRtuInterFrameTimeoutMs"), m_rtConfig.modbusInterFrameTimeoutMs);
auto s = plugin->createFramerSession(chunkIn.endpoint, params);
m_framerSessions.insert(
key, s ? std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>(std::move(s)) : nullptr);
it = m_framerSessions.find(key);
}
if (it != m_framerSessions.end() && it.value()) {
it.value()->feedChunk(chunkIn, completeFramesOut);
return;
}
}
} else if (m_passthroughFramer) {
}
if (m_passthroughFramer) {
m_passthroughFramer->feed(chunkIn, completeFramesOut);
}
}
// 功能 ：分配帧序列号