wirenboard / wb-mqtt-serial / 662

Committed 04 Jul 2025 09:27AM UTC coverage: 73.706% (-0.1%) from 73.835%

Build # 662

Build Type

push

github

Committed by

web-flow

Commit Message

Refactor device/LoadConfig RPC

Run Details

6427 of 9056 branches covered (70.97%)

0 of 114 new or added lines in 4 files covered. (0.0%)

8 existing lines in 3 files now uncovered.

12278 of 16658 relevant lines covered (73.71%)

306.66 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,
                                                      SerialClient ? SerialClient->GetPort() : TaskExecutor->GetPort(),
                                                      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
UNCOV 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
NEW 50	TRPCDeviceHelper::TRPCDeviceHelper(const Json::Value& request,	×
51	const TSerialDeviceFactory& deviceFactory,
52	PTemplateMap templates,
NEW 53	TSerialClientTaskRunner& serialClientTaskRunner)	×
54	{
NEW 55	auto params = serialClientTaskRunner.GetSerialClientParams(request);	×
NEW 56	SerialClient = params.SerialClient;	×
NEW 57	if (SerialClient == nullptr) {	×
NEW 58	TaskExecutor = serialClientTaskRunner.GetTaskExecutor(request);	×
59	}
NEW 60	if (params.Device == nullptr) {	×
NEW 61	DeviceTemplate = templates->GetTemplate(request["device_type"].asString());	×
62	auto config = std::make_shared<TDeviceConfig>("RPC Device",
NEW 63	request["slave_id"].asString(),	×
NEW 64	DeviceTemplate->GetProtocol());	×
NEW 65	if (DeviceTemplate->GetProtocol() == "modbus") {	×
NEW 66	config->MaxRegHole = Modbus::MAX_HOLE_CONTINUOUS_16_BIT_REGISTERS;	×
NEW 67	config->MaxBitHole = Modbus::MAX_HOLE_CONTINUOUS_1_BIT_REGISTERS;	×
NEW 68	config->MaxReadRegisters = Modbus::MAX_READ_REGISTERS;	×
69	}
NEW 70	ProtocolParams = deviceFactory.GetProtocolParams(DeviceTemplate->GetProtocol());	×
NEW 71	Device = ProtocolParams.factory->CreateDevice(DeviceTemplate->GetTemplate(),	×
72	config,
NEW 73	SerialClient ? SerialClient->GetPort() : TaskExecutor->GetPort(),	×
NEW 74	ProtocolParams.protocol);	×
75	} else {
NEW 76	Device = params.Device;	×
NEW 77	DeviceTemplate = templates->GetTemplate(Device->DeviceConfig()->DeviceType);	×
NEW 78	ProtocolParams = deviceFactory.GetProtocolParams(DeviceTemplate->GetProtocol());	×
NEW 79	DeviceFromConfig = true;	×
80	}
NEW 81	if (DeviceTemplate->WithSubdevices()) {	×
NEW 82	throw TRPCException("Device \"" + DeviceTemplate->Type + "\" is not supported by this RPC",	×
NEW 83	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
84	}
85	}
86
NEW 87	void TRPCDeviceHelper::RunTask(PSerialClientTask task)	×
88	{
NEW 89	if (SerialClient) {	×
NEW 90	SerialClient->AddTask(task);	×
91	} else {
NEW 92	TaskExecutor->AddTask(task);	×
93	}
94	}
95
UNCOV 96	TRPCDeviceHandler::TRPCDeviceHandler(const std::string& requestDeviceLoadConfigSchemaFilePath,	×
97	const std::string& requestDeviceProbeSchemaFilePath,
98	const TSerialDeviceFactory& deviceFactory,
99	PTemplateMap templates,
100	TSerialClientTaskRunner& serialClientTaskRunner,
101	TRPCDeviceParametersCache& parametersCache,
102	WBMQTT::PMqttRpcServer rpcServer)	×
103	: DeviceFactory(deviceFactory),
104	RequestDeviceLoadConfigSchema(LoadRPCRequestSchema(requestDeviceLoadConfigSchemaFilePath, "device/LoadConfig")),	×
105	RequestDeviceProbeSchema(LoadRPCRequestSchema(requestDeviceProbeSchemaFilePath, "device/Probe")),	×
106	Templates(templates),
107	SerialClientTaskRunner(serialClientTaskRunner),
108	ParametersCache(parametersCache)	×
109	{
110	rpcServer->RegisterAsyncMethod("device",	×
111	"LoadConfig",
112	std::bind(&TRPCDeviceHandler::LoadConfig,	×
113	this,	×
114	std::placeholders::_1,
115	std::placeholders::_2,
116	std::placeholders::_3));	×
117	rpcServer->RegisterAsyncMethod("device",	×
118	"Probe",
119	std::bind(&TRPCDeviceHandler::Probe,	×
120	this,	×
121	std::placeholders::_1,
122	std::placeholders::_2,
123	std::placeholders::_3));	×
124	}
125
126	void TRPCDeviceHandler::LoadConfig(const Json::Value& request,	×
127	WBMQTT::TMqttRpcServer::TResultCallback onResult,
128	WBMQTT::TMqttRpcServer::TErrorCallback onError)
129	{
130	ValidateRPCRequest(request, RequestDeviceLoadConfigSchema);	×
131	try {
NEW 132	auto helper = TRPCDeviceHelper(request, DeviceFactory, Templates, SerialClientTaskRunner);	×
133	auto rpcRequest = ParseRPCDeviceLoadConfigRequest(request,
134	helper.ProtocolParams,
135	helper.Device,
136	helper.DeviceTemplate,
137	helper.DeviceFromConfig,
138	ParametersCache,
139	onResult,
NEW 140	onError);	×
NEW 141	helper.RunTask(std::make_shared<TRPCDeviceLoadConfigSerialClientTask>(rpcRequest));	×
142	} catch (const TRPCException& e) {	×
143	ProcessException(e, onError);	×
144	}
145	}
146
147	void TRPCDeviceHandler::Probe(const Json::Value& request,	×
148	WBMQTT::TMqttRpcServer::TResultCallback onResult,
149	WBMQTT::TMqttRpcServer::TErrorCallback onError)
150	{
151	ValidateRPCRequest(request, RequestDeviceProbeSchema);	×
152	try {
153	SerialClientTaskRunner.RunTask(request,	×
154	std::make_shared<TRPCDeviceProbeSerialClientTask>(request, onResult, onError));	×
155	} catch (const TRPCException& e) {	×
156	ProcessException(e, onError);	×
157	}
158	}

wirenboard / wb-mqtt-serial / 662

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