wirenboard / wb-mqtt-serial / 666

Committed 08 Jul 2025 11:10AM UTC coverage: 73.854% (+0.1%) from 73.706%

Build # 666

Build Type

push

github

Committed by

web-flow

Commit Message

Port handling refactoring (#960)

Pass port by reference when needed instead of storing it in every device

Run Details

6444 of 9057 branches covered (71.15%)

526 of 700 new or added lines in 56 files covered. (75.14%)

6 existing lines in 5 files now uncovered.

12341 of 16710 relevant lines covered (73.85%)

305.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/rpc/rpc_device_handler.cpp

#include "rpc_device_handler.h"
#include "rpc_device_load_config_task.h"
#include "rpc_device_probe_task.h"
#include "rpc_helpers.h"

void TRPCDeviceParametersCache::RegisterCallbacks(PHandlerConfig handlerConfig)
{
    for (const auto& portConfig: handlerConfig->PortConfigs) {
        for (const auto& device: portConfig->Devices) {
            std::string id = GetId(*portConfig->Port, device->Device->DeviceConfig()->SlaveId);
            device->Device->AddOnConnectionStateChangedCallback([this, id](PSerialDevice device) {
                if (device->GetConnectionState() == TDeviceConnectionState::DISCONNECTED) {
                    Remove(id);
                }
            });
        }
    }
}

std::string TRPCDeviceParametersCache::GetId(const TPort& port, const std::string& slaveId) const
{
    return port.GetDescription(false) + ":" + slaveId;
}

void TRPCDeviceParametersCache::Add(const std::string& id, const Json::Value& value)
{
    std::unique_lock lock(Mutex);
    DeviceParameters[id] = value;
}

void TRPCDeviceParametersCache::Remove(const std::string& id)
{
    std::unique_lock lock(Mutex);
    DeviceParameters.erase(id);
}

bool TRPCDeviceParametersCache::Contains(const std::string& id) const
{
    std::unique_lock lock(Mutex);
    return DeviceParameters.find(id) != DeviceParameters.end();
}

const Json::Value& TRPCDeviceParametersCache::Get(const std::string& id, const Json::Value& defaultValue) const
{
    std::unique_lock lock(Mutex);
    auto it = DeviceParameters.find(id);
    return it != DeviceParameters.end() ? it->second : defaultValue;
};

TRPCDeviceHelper::TRPCDeviceHelper(const Json::Value& request,
                                   const TSerialDeviceFactory& deviceFactory,
                                   PTemplateMap templates,
                                   TSerialClientTaskRunner& serialClientTaskRunner)
{
    auto params = serialClientTaskRunner.GetSerialClientParams(request);
    SerialClient = params.SerialClient;
    if (SerialClient == nullptr) {
        TaskExecutor = serialClientTaskRunner.GetTaskExecutor(request);
    }
    if (params.Device == nullptr) {
        DeviceTemplate = templates->GetTemplate(request["device_type"].asString());
        auto config = std::make_shared<TDeviceConfig>("RPC Device",
                                                      request["slave_id"].asString(),
                                                      DeviceTemplate->GetProtocol());
        if (DeviceTemplate->GetProtocol() == "modbus") {
            config->MaxRegHole = Modbus::MAX_HOLE_CONTINUOUS_16_BIT_REGISTERS;
            config->MaxBitHole = Modbus::MAX_HOLE_CONTINUOUS_1_BIT_REGISTERS;
            config->MaxReadRegisters = Modbus::MAX_READ_REGISTERS;
        }
        ProtocolParams = deviceFactory.GetProtocolParams(DeviceTemplate->GetProtocol());
        Device = ProtocolParams.factory->CreateDevice(DeviceTemplate->GetTemplate(), config, ProtocolParams.protocol);
    } else {
        Device = params.Device;
        DeviceTemplate = templates->GetTemplate(Device->DeviceConfig()->DeviceType);
        ProtocolParams = deviceFactory.GetProtocolParams(DeviceTemplate->GetProtocol());
        DeviceFromConfig = true;
    }
    if (DeviceTemplate->WithSubdevices()) {
        throw TRPCException("Device \"" + DeviceTemplate->Type + "\" is not supported by this RPC",
                            TRPCResultCode::RPC_WRONG_PARAM_VALUE);
    }
}

void TRPCDeviceHelper::RunTask(PSerialClientTask task)
{
    if (SerialClient) {
        SerialClient->AddTask(task);
    } else {
        TaskExecutor->AddTask(task);
    }
}

TRPCDeviceHandler::TRPCDeviceHandler(const std::string& requestDeviceLoadConfigSchemaFilePath,
                                     const std::string& requestDeviceProbeSchemaFilePath,
                                     const TSerialDeviceFactory& deviceFactory,
                                     PTemplateMap templates,
                                     TSerialClientTaskRunner& serialClientTaskRunner,
                                     TRPCDeviceParametersCache& parametersCache,
                                     WBMQTT::PMqttRpcServer rpcServer)
    : DeviceFactory(deviceFactory),
      RequestDeviceLoadConfigSchema(LoadRPCRequestSchema(requestDeviceLoadConfigSchemaFilePath, "device/LoadConfig")),
      RequestDeviceProbeSchema(LoadRPCRequestSchema(requestDeviceProbeSchemaFilePath, "device/Probe")),
      Templates(templates),
      SerialClientTaskRunner(serialClientTaskRunner),
      ParametersCache(parametersCache)
{
    rpcServer->RegisterAsyncMethod("device",
                                   "LoadConfig",
                                   std::bind(&TRPCDeviceHandler::LoadConfig,
                                             this,
                                             std::placeholders::_1,
                                             std::placeholders::_2,
                                             std::placeholders::_3));
    rpcServer->RegisterAsyncMethod("device",
                                   "Probe",
                                   std::bind(&TRPCDeviceHandler::Probe,
                                             this,
                                             std::placeholders::_1,
                                             std::placeholders::_2,
                                             std::placeholders::_3));
}

void TRPCDeviceHandler::LoadConfig(const Json::Value& request,
                                   WBMQTT::TMqttRpcServer::TResultCallback onResult,
                                   WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    ValidateRPCRequest(request, RequestDeviceLoadConfigSchema);
    try {
        auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, SerialClientTaskRunner);
        auto rpcRequest = ParseRPCDeviceLoadConfigRequest(request,
                                                          helper.ProtocolParams,
                                                          helper.Device,
                                                          helper.DeviceTemplate,
                                                          helper.DeviceFromConfig,
                                                          ParametersCache,
                                                          onResult,
                                                          onError);
        helper.RunTask(std::make_shared<TRPCDeviceLoadConfigSerialClientTask>(rpcRequest));
    } catch (const TRPCException& e) {
        ProcessException(e, onError);
    }
}

void TRPCDeviceHandler::Probe(const Json::Value& request,
                              WBMQTT::TMqttRpcServer::TResultCallback onResult,
                              WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    ValidateRPCRequest(request, RequestDeviceProbeSchema);
    try {
        SerialClientTaskRunner.RunTask(request,
                                       std::make_shared<TRPCDeviceProbeSerialClientTask>(request, onResult, onError));
    } catch (const TRPCException& e) {
        ProcessException(e, onError);
    }
}

1	#include "rpc_device_handler.h"
2	#include "rpc_device_load_config_task.h"
3	#include "rpc_device_probe_task.h"
4	#include "rpc_helpers.h"
5
6	void TRPCDeviceParametersCache::RegisterCallbacks(PHandlerConfig handlerConfig)	×
7	{
8	for (const auto& portConfig: handlerConfig->PortConfigs) {	×
9	for (const auto& device: portConfig->Devices) {	×
10	std::string id = GetId(*portConfig->Port, device->Device->DeviceConfig()->SlaveId);	×
11	device->Device->AddOnConnectionStateChangedCallback([this, id](PSerialDevice device) {	×
12	if (device->GetConnectionState() == TDeviceConnectionState::DISCONNECTED) {	×
13	Remove(id);	×
14	}
15	});	×
16	}
17	}
18	}
19
20	std::string TRPCDeviceParametersCache::GetId(const TPort& port, const std::string& slaveId) const	×
21	{
22	return port.GetDescription(false) + ":" + slaveId;	×
23	}
24
25	void TRPCDeviceParametersCache::Add(const std::string& id, const Json::Value& value)	×
26	{
27	std::unique_lock lock(Mutex);	×
28	DeviceParameters[id] = value;	×
29	}
30
31	void TRPCDeviceParametersCache::Remove(const std::string& id)	×
32	{
33	std::unique_lock lock(Mutex);	×
34	DeviceParameters.erase(id);	×
35	}
36
37	bool TRPCDeviceParametersCache::Contains(const std::string& id) const	×
38	{
39	std::unique_lock lock(Mutex);	×
40	return DeviceParameters.find(id) != DeviceParameters.end();	×
41	}
42
43	const Json::Value& TRPCDeviceParametersCache::Get(const std::string& id, const Json::Value& defaultValue) const	×
44	{
45	std::unique_lock lock(Mutex);	×
46	auto it = DeviceParameters.find(id);	×
47	return it != DeviceParameters.end() ? it->second : defaultValue;	×
48	};
49
50	TRPCDeviceHelper::TRPCDeviceHelper(const Json::Value& request,	×
51	const TSerialDeviceFactory& deviceFactory,
52	PTemplateMap templates,
53	TSerialClientTaskRunner& serialClientTaskRunner)	×
54	{
55	auto params = serialClientTaskRunner.GetSerialClientParams(request);	×
56	SerialClient = params.SerialClient;	×
57	if (SerialClient == nullptr) {	×
58	TaskExecutor = serialClientTaskRunner.GetTaskExecutor(request);	×
59	}
60	if (params.Device == nullptr) {	×
61	DeviceTemplate = templates->GetTemplate(request["device_type"].asString());	×
62	auto config = std::make_shared<TDeviceConfig>("RPC Device",
63	request["slave_id"].asString(),	×
64	DeviceTemplate->GetProtocol());	×
65	if (DeviceTemplate->GetProtocol() == "modbus") {	×
66	config->MaxRegHole = Modbus::MAX_HOLE_CONTINUOUS_16_BIT_REGISTERS;	×
67	config->MaxBitHole = Modbus::MAX_HOLE_CONTINUOUS_1_BIT_REGISTERS;	×
68	config->MaxReadRegisters = Modbus::MAX_READ_REGISTERS;	×
69	}
70	ProtocolParams = deviceFactory.GetProtocolParams(DeviceTemplate->GetProtocol());	×
NEW 71	Device = ProtocolParams.factory->CreateDevice(DeviceTemplate->GetTemplate(), config, ProtocolParams.protocol);	×
72	} else {
73	Device = params.Device;	×
74	DeviceTemplate = templates->GetTemplate(Device->DeviceConfig()->DeviceType);	×
75	ProtocolParams = deviceFactory.GetProtocolParams(DeviceTemplate->GetProtocol());	×
76	DeviceFromConfig = true;	×
77	}
78	if (DeviceTemplate->WithSubdevices()) {	×
79	throw TRPCException("Device \"" + DeviceTemplate->Type + "\" is not supported by this RPC",	×
80	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
81	}
82	}
83
84	void TRPCDeviceHelper::RunTask(PSerialClientTask task)	×
85	{
86	if (SerialClient) {	×
87	SerialClient->AddTask(task);	×
88	} else {
89	TaskExecutor->AddTask(task);	×
90	}
91	}
92
93	TRPCDeviceHandler::TRPCDeviceHandler(const std::string& requestDeviceLoadConfigSchemaFilePath,	×
94	const std::string& requestDeviceProbeSchemaFilePath,
95	const TSerialDeviceFactory& deviceFactory,
96	PTemplateMap templates,
97	TSerialClientTaskRunner& serialClientTaskRunner,
98	TRPCDeviceParametersCache& parametersCache,
99	WBMQTT::PMqttRpcServer rpcServer)	×
100	: DeviceFactory(deviceFactory),
101	RequestDeviceLoadConfigSchema(LoadRPCRequestSchema(requestDeviceLoadConfigSchemaFilePath, "device/LoadConfig")),	×
102	RequestDeviceProbeSchema(LoadRPCRequestSchema(requestDeviceProbeSchemaFilePath, "device/Probe")),	×
103	Templates(templates),
104	SerialClientTaskRunner(serialClientTaskRunner),
105	ParametersCache(parametersCache)	×
106	{
107	rpcServer->RegisterAsyncMethod("device",	×
108	"LoadConfig",
109	std::bind(&TRPCDeviceHandler::LoadConfig,	×
110	this,	×
111	std::placeholders::_1,
112	std::placeholders::_2,
113	std::placeholders::_3));	×
114	rpcServer->RegisterAsyncMethod("device",	×
115	"Probe",
116	std::bind(&TRPCDeviceHandler::Probe,	×
117	this,	×
118	std::placeholders::_1,
119	std::placeholders::_2,
120	std::placeholders::_3));	×
121	}
122
123	void TRPCDeviceHandler::LoadConfig(const Json::Value& request,	×
124	WBMQTT::TMqttRpcServer::TResultCallback onResult,
125	WBMQTT::TMqttRpcServer::TErrorCallback onError)
126	{
127	ValidateRPCRequest(request, RequestDeviceLoadConfigSchema);	×
128	try {
129	auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, SerialClientTaskRunner);	×
130	auto rpcRequest = ParseRPCDeviceLoadConfigRequest(request,
131	helper.ProtocolParams,
132	helper.Device,
133	helper.DeviceTemplate,
134	helper.DeviceFromConfig,
135	ParametersCache,
136	onResult,
137	onError);	×
138	helper.RunTask(std::make_shared<TRPCDeviceLoadConfigSerialClientTask>(rpcRequest));	×
139	} catch (const TRPCException& e) {	×
140	ProcessException(e, onError);	×
141	}
142	}
143
144	void TRPCDeviceHandler::Probe(const Json::Value& request,	×
145	WBMQTT::TMqttRpcServer::TResultCallback onResult,
146	WBMQTT::TMqttRpcServer::TErrorCallback onError)
147	{
148	ValidateRPCRequest(request, RequestDeviceProbeSchema);	×
149	try {
150	SerialClientTaskRunner.RunTask(request,	×
151	std::make_shared<TRPCDeviceProbeSerialClientTask>(request, onResult, onError));	×
152	} catch (const TRPCException& e) {	×
153	ProcessException(e, onError);	×
154	}
155	}

wirenboard / wb-mqtt-serial / 666

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous