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_port_load_modbus_serial_client_task.cpp

#include "rpc_port_load_modbus_serial_client_task.h"
#include "modbus_base.h"
#include "rpc_port_handler.h"
#include "serial_exc.h"
#include "serial_port.h"

template<> bool inline WBMQTT::JSON::Is<uint8_t>(const Json::Value& value)
{
    return value.isUInt();
}

template<> inline uint8_t WBMQTT::JSON::As<uint8_t>(const Json::Value& value)
{
    return value.asUInt() & 0xFF;
}

template<> bool inline WBMQTT::JSON::Is<uint16_t>(const Json::Value& value)
{
    return value.isUInt();
}

template<> inline uint16_t WBMQTT::JSON::As<uint16_t>(const Json::Value& value)
{
    return value.asUInt() & 0xFFFF;
}

template<> bool inline WBMQTT::JSON::Is<Modbus::EFunction>(const Json::Value& value)
{
    if (!value.isUInt()) {
        return false;
    }
    auto fun = value.asUInt();
    return Modbus::IsSupportedFunction(fun & 0xFF);
}

template<> inline Modbus::EFunction WBMQTT::JSON::As<Modbus::EFunction>(const Json::Value& value)
{
    return static_cast<Modbus::EFunction>(value.asUInt() & 0xFF);
}

TRPCPortLoadModbusRequest::TRPCPortLoadModbusRequest(TRPCDeviceParametersCache& parametersCache)
    : ParametersCache(parametersCache)
{}

PRPCPortLoadModbusRequest ParseRPCPortLoadModbusRequest(const Json::Value& request,
                                                        TRPCDeviceParametersCache& parametersCache)
{
    PRPCPortLoadModbusRequest RPCRequest = std::make_shared<TRPCPortLoadModbusRequest>(parametersCache);

    try {
        ParseRPCPortLoadRequest(request, *RPCRequest);
        WBMQTT::JSON::Get(request, "slave_id", RPCRequest->SlaveId);
        WBMQTT::JSON::Get(request, "address", RPCRequest->Address);
        WBMQTT::JSON::Get(request, "count", RPCRequest->Count);
        WBMQTT::JSON::Get(request, "function", RPCRequest->Function);
    } catch (const std::runtime_error& e) {
        throw TRPCException(e.what(), TRPCResultCode::RPC_WRONG_PARAM_VALUE);
    }

    return RPCRequest;
}

void ExecRPCPortLoadModbusRequest(TPort& port, PRPCPortLoadModbusRequest rpcRequest)
{
    try {
        port.CheckPortOpen();
        port.SkipNoise();
        port.SleepSinceLastInteraction(rpcRequest->FrameTimeout);
        Modbus::TModbusRTUTraits traits;
        auto pdu = Modbus::MakePDU(rpcRequest->Function, rpcRequest->Address, rpcRequest->Count, rpcRequest->Message);
        auto responsePduSize = Modbus::CalcResponsePDUSize(rpcRequest->Function, rpcRequest->Count);
        auto res = traits.Transaction(port,
                                      rpcRequest->SlaveId,
                                      pdu,
                                      responsePduSize,
                                      rpcRequest->ResponseTimeout,
                                      rpcRequest->FrameTimeout);
        auto response = Modbus::ExtractResponseData(rpcRequest->Function, res.Pdu);

        if (rpcRequest->OnResult) {
            Json::Value replyJSON;
            replyJSON["response"] = FormatResponse(response, rpcRequest->Format);
            rpcRequest->OnResult(replyJSON);
        }

        if (rpcRequest->Function == Modbus::EFunction::FN_WRITE_SINGLE_COIL ||
            rpcRequest->Function == Modbus::EFunction::FN_WRITE_SINGLE_REGISTER ||
            rpcRequest->Function == Modbus::EFunction::FN_WRITE_MULTIPLE_COILS ||
            rpcRequest->Function == Modbus::EFunction::FN_WRITE_MULTIPLE_REGISTERS)
        {
            std::string id = rpcRequest->ParametersCache.GetId(port, std::to_string(rpcRequest->SlaveId));
            rpcRequest->ParametersCache.Remove(id);
        }
    } catch (const Modbus::TModbusExceptionError& error) {
        Json::Value replyJSON;
        replyJSON["exception"]["code"] = error.GetExceptionCode();
        replyJSON["exception"]["msg"] = error.what();
        rpcRequest->OnResult(replyJSON);
    } catch (const TResponseTimeoutException& error) {
        if (rpcRequest->OnError) {
            rpcRequest->OnError(WBMQTT::E_RPC_REQUEST_TIMEOUT, error.what());
        }
    }
}

TRPCPortLoadModbusSerialClientTask::TRPCPortLoadModbusSerialClientTask(const Json::Value& request,
                                                                       WBMQTT::TMqttRpcServer::TResultCallback onResult,
                                                                       WBMQTT::TMqttRpcServer::TErrorCallback onError,
                                                                       TRPCDeviceParametersCache& parametersCache)
    : Request(ParseRPCPortLoadModbusRequest(request, parametersCache))
{
    Request->OnResult = onResult;
    Request->OnError = onError;
    ExpireTime = std::chrono::steady_clock::now() + Request->TotalTimeout;
}

ISerialClientTask::TRunResult TRPCPortLoadModbusSerialClientTask::Run(
    PPort port,
    TSerialClientDeviceAccessHandler& lastAccessedDevice,
    const std::list<PSerialDevice>& polledDevices)
{
    if (std::chrono::steady_clock::now() > ExpireTime) {
        if (Request->OnError) {
            Request->OnError(WBMQTT::E_RPC_REQUEST_TIMEOUT, "RPC request timeout");
        }
        return ISerialClientTask::TRunResult::OK;
    }

    try {
        if (!port->IsOpen()) {
            port->Open();
        }
        lastAccessedDevice.PrepareToAccess(*port, nullptr);
        TSerialPortSettingsGuard settingsGuard(port, Request->SerialPortSettings);
        ExecRPCPortLoadModbusRequest(*port, Request);
    } catch (const std::exception& error) {
        if (Request->OnError) {
            Request->OnError(WBMQTT::E_RPC_SERVER_ERROR, std::string("Port IO error: ") + error.what());
        }
    }
    return ISerialClientTask::TRunResult::OK;
}

1	#include "rpc_port_load_modbus_serial_client_task.h"
2	#include "modbus_base.h"
3	#include "rpc_port_handler.h"
4	#include "serial_exc.h"
5	#include "serial_port.h"
6
7	template<> bool inline WBMQTT::JSON::Is<uint8_t>(const Json::Value& value)	×
8	{
9	return value.isUInt();	×
10	}
11
12	template<> inline uint8_t WBMQTT::JSON::As<uint8_t>(const Json::Value& value)	×
13	{
14	return value.asUInt() & 0xFF;	×
15	}
16
17	template<> bool inline WBMQTT::JSON::Is<uint16_t>(const Json::Value& value)	×
18	{
19	return value.isUInt();	×
20	}
21
22	template<> inline uint16_t WBMQTT::JSON::As<uint16_t>(const Json::Value& value)	×
23	{
24	return value.asUInt() & 0xFFFF;	×
25	}
26
27	template<> bool inline WBMQTT::JSON::Is<Modbus::EFunction>(const Json::Value& value)	×
28	{
29	if (!value.isUInt()) {	×
30	return false;	×
31	}
32	auto fun = value.asUInt();	×
33	return Modbus::IsSupportedFunction(fun & 0xFF);	×
34	}
35
36	template<> inline Modbus::EFunction WBMQTT::JSON::As<Modbus::EFunction>(const Json::Value& value)	×
37	{
38	return static_cast<Modbus::EFunction>(value.asUInt() & 0xFF);	×
39	}
40
41	TRPCPortLoadModbusRequest::TRPCPortLoadModbusRequest(TRPCDeviceParametersCache& parametersCache)	×
42	: ParametersCache(parametersCache)	×
43	{}
44
45	PRPCPortLoadModbusRequest ParseRPCPortLoadModbusRequest(const Json::Value& request,	×
46	TRPCDeviceParametersCache& parametersCache)
47	{
48	PRPCPortLoadModbusRequest RPCRequest = std::make_shared<TRPCPortLoadModbusRequest>(parametersCache);	×
49
50	try {
51	ParseRPCPortLoadRequest(request, *RPCRequest);	×
52	WBMQTT::JSON::Get(request, "slave_id", RPCRequest->SlaveId);	×
53	WBMQTT::JSON::Get(request, "address", RPCRequest->Address);	×
54	WBMQTT::JSON::Get(request, "count", RPCRequest->Count);	×
55	WBMQTT::JSON::Get(request, "function", RPCRequest->Function);	×
56	} catch (const std::runtime_error& e) {	×
57	throw TRPCException(e.what(), TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
58	}
59
60	return RPCRequest;	×
61	}
62
63	void ExecRPCPortLoadModbusRequest(TPort& port, PRPCPortLoadModbusRequest rpcRequest)	×
64	{
65	try {
66	port.CheckPortOpen();	×
67	port.SkipNoise();	×
68	port.SleepSinceLastInteraction(rpcRequest->FrameTimeout);	×
69	Modbus::TModbusRTUTraits traits;	×
70	auto pdu = Modbus::MakePDU(rpcRequest->Function, rpcRequest->Address, rpcRequest->Count, rpcRequest->Message);	×
71	auto responsePduSize = Modbus::CalcResponsePDUSize(rpcRequest->Function, rpcRequest->Count);	×
72	auto res = traits.Transaction(port,
73	rpcRequest->SlaveId,	×
74	pdu,
75	responsePduSize,
76	rpcRequest->ResponseTimeout,	×
77	rpcRequest->FrameTimeout);	×
78	auto response = Modbus::ExtractResponseData(rpcRequest->Function, res.Pdu);	×
79
80	if (rpcRequest->OnResult) {	×
81	Json::Value replyJSON;	×
82	replyJSON["response"] = FormatResponse(response, rpcRequest->Format);	×
83	rpcRequest->OnResult(replyJSON);	×
84	}
85
86	if (rpcRequest->Function == Modbus::EFunction::FN_WRITE_SINGLE_COIL \|\|	×
87	rpcRequest->Function == Modbus::EFunction::FN_WRITE_SINGLE_REGISTER \|\|	×
88	rpcRequest->Function == Modbus::EFunction::FN_WRITE_MULTIPLE_COILS \|\|	×
89	rpcRequest->Function == Modbus::EFunction::FN_WRITE_MULTIPLE_REGISTERS)	×
90	{
91	std::string id = rpcRequest->ParametersCache.GetId(port, std::to_string(rpcRequest->SlaveId));	×
92	rpcRequest->ParametersCache.Remove(id);	×
93	}
94	} catch (const Modbus::TModbusExceptionError& error) {	×
95	Json::Value replyJSON;	×
96	replyJSON["exception"]["code"] = error.GetExceptionCode();	×
97	replyJSON["exception"]["msg"] = error.what();	×
98	rpcRequest->OnResult(replyJSON);	×
99	} catch (const TResponseTimeoutException& error) {	×
100	if (rpcRequest->OnError) {	×
101	rpcRequest->OnError(WBMQTT::E_RPC_REQUEST_TIMEOUT, error.what());	×
102	}
103	}
104	}
105
106	TRPCPortLoadModbusSerialClientTask::TRPCPortLoadModbusSerialClientTask(const Json::Value& request,	×
107	WBMQTT::TMqttRpcServer::TResultCallback onResult,
108	WBMQTT::TMqttRpcServer::TErrorCallback onError,
109	TRPCDeviceParametersCache& parametersCache)	×
110	: Request(ParseRPCPortLoadModbusRequest(request, parametersCache))	×
111	{
112	Request->OnResult = onResult;	×
113	Request->OnError = onError;	×
114	ExpireTime = std::chrono::steady_clock::now() + Request->TotalTimeout;	×
115	}
116
117	ISerialClientTask::TRunResult TRPCPortLoadModbusSerialClientTask::Run(	×
118	PPort port,
119	TSerialClientDeviceAccessHandler& lastAccessedDevice,
120	const std::list<PSerialDevice>& polledDevices)
121	{
122	if (std::chrono::steady_clock::now() > ExpireTime) {	×
123	if (Request->OnError) {	×
124	Request->OnError(WBMQTT::E_RPC_REQUEST_TIMEOUT, "RPC request timeout");	×
125	}
126	return ISerialClientTask::TRunResult::OK;	×
127	}
128
129	try {
130	if (!port->IsOpen()) {	×
131	port->Open();	×
132	}
NEW 133	lastAccessedDevice.PrepareToAccess(*port, nullptr);	×
134	TSerialPortSettingsGuard settingsGuard(port, Request->SerialPortSettings);	×
135	ExecRPCPortLoadModbusRequest(*port, Request);	×
136	} catch (const std::exception& error) {	×
137	if (Request->OnError) {	×
138	Request->OnError(WBMQTT::E_RPC_SERVER_ERROR, std::string("Port IO error: ") + error.what());	×
139	}
140	}
141	return ISerialClientTask::TRunResult::OK;	×
142	}

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