wirenboard / wb-mqtt-serial / 967

Committed 05 Oct 2026 09:41AM UTC coverage: 85.672% (+2.4%) from 83.268%

Build # 967

Build Type

push

github

Committed by

web-flow

Commit Message

Add config/Save RPC (#1273)

The RPC validates a configuration, writes it to the configuration file and
restarts the polling in-process. Port clients are stopped first, requests
arriving meanwhile get config-busy and queued ones get request-cancelled.

Coverage Stats

12445 of 14263 branches covered (87.25%)

1432 of 1623 new or added lines in 41 files covered. (88.23%)

8 existing lines in 4 files now uncovered.

21065 of 24588 relevant lines covered (85.67%)

1528.89 hits per line

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

80.92

/src/rpc/rpc_device_load_task.cpp

#include "rpc_device_load_task.h"
#include "config_merge_template.h"
#include "port/serial_port.h"
#include "rpc_helpers.h"

namespace
{
    void ReadRegisters(PPort port, PRPCDeviceLoadRequest rpcRequest, TRPCRegisterList& registerList, Json::Value& data)
    {
        ReadRegisterList(*port, rpcRequest->Device, registerList);
        for (const auto& item: registerList) {
            data[item.Id] = RegisterGotValue(item)
                                ? RawValueToJSON(*item.Register->GetConfig(), item.Register->GetValue())
                                : UNSUPPORTED_VALUE;
        }
        MarkUnsupportedRegisterItems(*port, *rpcRequest, registerList, &data);
    }

    void ExecRPCRequest(PPort port, PRPCDeviceLoadRequest rpcRequest)
    {
        if (!rpcRequest->OnResult) {
            return;
        }

        PrepareSession(*port, rpcRequest->Device);

        Json::Value result(Json::objectValue);
        try {
            // Step 1: Read parameters that are referenced by conditions of the requested items.
            Json::Value conditionParamValues(Json::objectValue);
            auto condParamRegList = rpcRequest->GetConditionParametersRegisterList();
            ReadRegisterList(*port, rpcRequest->Device, condParamRegList);
            MarkUnsupportedRegisterItems(*port, *rpcRequest, condParamRegList);
            MergeRegisterListValues(condParamRegList, conditionParamValues);

            // Step 2: Read channels, filtering by condition using actual parameter values
            Json::Value readonlyList(Json::arrayValue);
            Json::Value channelData(Json::objectValue);
            auto channelRegList = rpcRequest->GetChannelsRegisterList(conditionParamValues);
            ReadRegisters(port, rpcRequest, channelRegList, channelData);
            for (const auto& item: channelRegList) {
                if (item.Register->GetConfig()->AccessType == TRegisterConfig::EAccessType::READ_ONLY) {
                    readonlyList.append(item.Id);
                }
            }
            if (!channelData.empty()) {
                result["channels"] = channelData;
            }

            // Step 3: Read explicitly requested parameters, reusing values already
            // read for condition evaluation to avoid duplicate Modbus reads
            Json::Value paramData(Json::objectValue);
            auto paramRegList = rpcRequest->GetParametersRegisterList(conditionParamValues, &paramData);
            ReadRegisters(port, rpcRequest, paramRegList, paramData);
            if (!paramData.empty()) {
                result["parameters"] = paramData;
            }

            if (!readonlyList.empty()) {
                result["readonly"] = readonlyList;
            }
        } catch (...) {
            EndSession(*port, rpcRequest->Device);
            throw;
        }
        EndSession(*port, rpcRequest->Device);

        rpcRequest->OnResult(result);
    }
} // namespace

TRPCDeviceLoadRequest::TRPCDeviceLoadRequest(const TDeviceProtocolParams& protocolParams,
                                             PSerialDevice device,
                                             PDeviceTemplate deviceTemplate,
                                             bool deviceFromConfig)
    : TRPCDeviceRequest(protocolParams, device, deviceTemplate, deviceFromConfig)
{}

void TRPCDeviceLoadRequest::ParseRequestItems(const Json::Value& items, std::set<std::string>& list)
{
    for (const auto& item: items) {
        list.insert(item.asString());
    }
}

TRPCRegisterList TRPCDeviceLoadRequest::GetConditionParametersRegisterList()
{
    // Collect the parameter names referenced by conditions of the requested channels and parameters,
    // an empty channel list requests all channels
    std::set<std::string> neededParams;
    Expressions::TExpressionsCache exprCache;
    auto allChannels = Channels.empty();
    for (const auto& ch: DeviceTemplate->GetTemplate()["channels"]) {
        if (ch["address"].isNull()) { // write only channel
            continue;
        }
        if (!allChannels && !Channels.count(ch["name"].asString())) {
            continue;
        }
        auto deps = Expressions::GetDependencies(ch["condition"].asString(), exprCache);
        neededParams.insert(deps.begin(), deps.end());
    }
    const auto& params = DeviceTemplate->GetTemplate()["parameters"];
    for (auto it = params.begin(); it != params.end(); ++it) {
        const auto& item = *it;
        if (item["address"].isNull()) {
            continue;
        }
        auto id = params.isObject() ? it.key().asString() : item["id"].asString();
        if (!Parameters.count(id)) {
            continue;
        }
        auto deps = Expressions::GetDependencies(item["condition"].asString(), exprCache);
        neededParams.insert(deps.begin(), deps.end());
    }
    if (neededParams.empty()) {
        return {};
    }
    return CreateParametersRegisterList(ProtocolParams, Device, params, Json::Value(), neededParams);
}

TRPCRegisterList TRPCDeviceLoadRequest::GetChannelsRegisterList(const Json::Value& conditionParams)
{
    auto notFound = Channels;
    auto allChannels = Channels.empty();
    Json::Value items(Json::arrayValue);
    for (const auto& item: DeviceTemplate->GetTemplate()["channels"]) {
        if (item["address"].isNull()) { // write only channel
            continue;
        }
        auto id = item["name"].asString();
        if (allChannels || Channels.count(id)) {
            auto channel = item;
            channel["id"] = id;
            items.append(channel);
            notFound.erase(id);
        }
    }
    if (!notFound.empty()) {
        throw TRPCException("Channel \"" + *notFound.begin() + "\" is write only or not found in \"" +
                                DeviceTemplate->Type + "\" device template",
                            TRPCResultCode::RPC_WRONG_PARAM_VALUE);
    }

    // Filter channels by condition using actual device parameter values,
    // conditions over parameters without a value are evaluated with an undefined value
    if (!conditionParams.isNull()) {
        Expressions::TExpressionsCache cache;
        TJsonParams exprParams(conditionParams);
        Json::Value filtered(Json::arrayValue);
        for (const auto& item: items) {
            if (CheckCondition(item, exprParams, &cache)) {
                filtered.append(item);
            }
        }
        items = filtered;
    }

    return CreateChannelsRegisterList(ProtocolParams, Device, items);
}

TRPCRegisterList TRPCDeviceLoadRequest::GetParametersRegisterList(const Json::Value& conditionParams, Json::Value* data)
{
    const auto& params = DeviceTemplate->GetTemplate()["parameters"];
    auto notFound = Parameters;
    Expressions::TExpressionsCache cache;
    TJsonParams exprParams(conditionParams);
    TActiveParameterDeclarations matched;
    Json::Value items(Json::arrayValue);
    for (auto it = params.begin(); it != params.end(); ++it) {
        auto item = *it;
        if (item["address"].isNull()) {
            continue;
        }
        auto id = params.isObject() ? it.key().asString() : item["id"].asString();
        if (!Parameters.count(id)) {
            continue;
        }
        notFound.erase(id);
        if (!conditionParams.isNull()) {
            // conditions over parameters without a value are evaluated with an undefined value
            if (!CheckCondition(item, exprParams, &cache)) {
                continue;
            }
            // A chain of fw variants is not ambiguous, CreateParametersRegisterList merges it into one register.
            // Anything else is a template error, config validation reports it as a duplicate definition
            if (!matched.Add(id, *it)) {
                throw TRPCException("Parameter \"" + id +
                                        "\" is ambiguous: several declarations match the condition parameter values",
                                    TRPCResultCode::RPC_WRONG_PARAM_VALUE);
            }
        }
        if (conditionParams.isMember(id)) {
            // already read for condition evaluation, declarations of such a parameter read identically
            if (data != nullptr) {
                (*data)[id] = conditionParams[id];
            }
            continue;
        }
        if (params.isObject()) {
            item["id"] = id;
        }
        items.append(item);
    }
    if (!notFound.empty()) {
        throw TRPCException("Parameter \"" + *notFound.begin() + "\" is write only or not found in \"" +
                                DeviceTemplate->Type + "\" device template",
                            TRPCResultCode::RPC_WRONG_PARAM_VALUE);
    }

    return CreateParametersRegisterList(ProtocolParams, Device, items);
}

PRPCDeviceLoadRequest ParseRPCDeviceLoadRequest(const Json::Value& request,
                                                const TDeviceProtocolParams& protocolParams,
                                                PSerialDevice device,
                                                PDeviceTemplate deviceTemplate,
                                                bool deviceFromConfig,
                                                WBMQTT::TMqttRpcServer::TResultCallback onResult,
                                                WBMQTT::TMqttRpcServer::TErrorCallback onError)
{
    auto res = std::make_shared<TRPCDeviceLoadRequest>(protocolParams, device, deviceTemplate, deviceFromConfig);
    res->ParseSettings(request, onResult, onError);
    res->ParseRequestItems(request["channels"], res->Channels);
    res->ParseRequestItems(request["parameters"], res->Parameters);
    return res;
}

TRPCDeviceLoadSerialClientTask::TRPCDeviceLoadSerialClientTask(PRPCDeviceLoadRequest request): Request(request)
{
    ExpireTime = std::chrono::steady_clock::now() + Request->TotalTimeout;
}

ISerialClientTask::TRunResult TRPCDeviceLoadSerialClientTask::Run(PFeaturePort port,
                                                                  TSerialClientDeviceAccessHandler& lastAccessedDevice,
                                                                  const std::list<PSerialDevice>& polledDevices)
{
    if (IsCancelled()) {
        AnswerRequestCancelled(Request->OnError);
        return ISerialClientTask::TRunResult::OK;
    }
    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);
        if (!Request->DeviceFromConfig) {
            TSerialPortSettingsGuard settingsGuard(port, Request->SerialPortSettings);
            ExecRPCRequest(port, Request);
        } else {
            ExecRPCRequest(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_device_load_task.h"
2	#include "config_merge_template.h"
3	#include "port/serial_port.h"
4	#include "rpc_helpers.h"
5
6	namespace
7	{
8	void ReadRegisters(PPort port, PRPCDeviceLoadRequest rpcRequest, TRPCRegisterList& registerList, Json::Value& data)	4 ✔
9	{
10	ReadRegisterList(*port, rpcRequest->Device, registerList);	4 ✔
11	for (const auto& item: registerList) {	6 ✔
12	data[item.Id] = RegisterGotValue(item)	4 ✔
13	? RawValueToJSON(*item.Register->GetConfig(), item.Register->GetValue())	4 ✔
14	: UNSUPPORTED_VALUE;	2 ✔
15	}
16	MarkUnsupportedRegisterItems(port, rpcRequest, registerList, &data);	4 ✔
17	}	4 ✔
18
19	void ExecRPCRequest(PPort port, PRPCDeviceLoadRequest rpcRequest)	2 ✔
20	{
21	if (!rpcRequest->OnResult) {	2 ✔
22	return;	×
23	}
24
25	PrepareSession(*port, rpcRequest->Device);	2 ✔
26
27	Json::Value result(Json::objectValue);	2 ✔
28	try {
29	// Step 1: Read parameters that are referenced by conditions of the requested items.
30	Json::Value conditionParamValues(Json::objectValue);	2 ✔
31	auto condParamRegList = rpcRequest->GetConditionParametersRegisterList();	2 ✔
32	ReadRegisterList(*port, rpcRequest->Device, condParamRegList);	2 ✔
33	MarkUnsupportedRegisterItems(port, rpcRequest, condParamRegList);	2 ✔
34	MergeRegisterListValues(condParamRegList, conditionParamValues);	2 ✔
35
36	// Step 2: Read channels, filtering by condition using actual parameter values
37	Json::Value readonlyList(Json::arrayValue);	2 ✔
38	Json::Value channelData(Json::objectValue);	2 ✔
39	auto channelRegList = rpcRequest->GetChannelsRegisterList(conditionParamValues);	2 ✔
40	ReadRegisters(port, rpcRequest, channelRegList, channelData);	2 ✔
41	for (const auto& item: channelRegList) {	4 ✔
42	if (item.Register->GetConfig()->AccessType == TRegisterConfig::EAccessType::READ_ONLY) {	2 ✔
43	readonlyList.append(item.Id);	2 ✔
44	}
45	}
46	if (!channelData.empty()) {	2 ✔
47	result["channels"] = channelData;	1 ✔
48	}
49
50	// Step 3: Read explicitly requested parameters, reusing values already
51	// read for condition evaluation to avoid duplicate Modbus reads
52	Json::Value paramData(Json::objectValue);	2 ✔
53	auto paramRegList = rpcRequest->GetParametersRegisterList(conditionParamValues, &paramData);	2 ✔
54	ReadRegisters(port, rpcRequest, paramRegList, paramData);	2 ✔
55	if (!paramData.empty()) {	2 ✔
56	result["parameters"] = paramData;	×
57	}
58
59	if (!readonlyList.empty()) {	2 ✔
60	result["readonly"] = readonlyList;	1 ✔
61	}
62	} catch (...) {	2 ✔
63	EndSession(*port, rpcRequest->Device);	×
64	throw;	×
65	}	×
66	EndSession(*port, rpcRequest->Device);	2 ✔
67
68	rpcRequest->OnResult(result);	2 ✔
69	}	2 ✔
70	} // namespace
71
72	TRPCDeviceLoadRequest::TRPCDeviceLoadRequest(const TDeviceProtocolParams& protocolParams,	24 ✔
73	PSerialDevice device,
74	PDeviceTemplate deviceTemplate,
75	bool deviceFromConfig)	24 ✔
76	: TRPCDeviceRequest(protocolParams, device, deviceTemplate, deviceFromConfig)	24 ✔
77	{}	24 ✔
78
79	void TRPCDeviceLoadRequest::ParseRequestItems(const Json::Value& items, std::set<std::string>& list)	×
80	{
81	for (const auto& item: items) {	×
82	list.insert(item.asString());	×
83	}
84	}	×
85
86	TRPCRegisterList TRPCDeviceLoadRequest::GetConditionParametersRegisterList()	7 ✔
87	{
88	// Collect the parameter names referenced by conditions of the requested channels and parameters,
89	// an empty channel list requests all channels
90	std::set<std::string> neededParams;	7 ✔
91	Expressions::TExpressionsCache exprCache;	7 ✔
92	auto allChannels = Channels.empty();	7 ✔
93	for (const auto& ch: DeviceTemplate->GetTemplate()["channels"]) {	87 ✔
94	if (ch["address"].isNull()) { // write only channel	40 ✔
95	continue;	26 ✔
96	}
97	if (!allChannels && !Channels.count(ch["name"].asString())) {	27 ✔
98	continue;	13 ✔
99	}
100	auto deps = Expressions::GetDependencies(ch["condition"].asString(), exprCache);	14 ✔
101	neededParams.insert(deps.begin(), deps.end());	14 ✔
102	}	14 ✔
103	const auto& params = DeviceTemplate->GetTemplate()["parameters"];	7 ✔
104	for (auto it = params.begin(); it != params.end(); ++it) {	77 ✔
105	const auto& item = *it;	35 ✔
106	if (item["address"].isNull()) {	35 ✔
107	continue;	32 ✔
108	}
109	auto id = params.isObject() ? it.key().asString() : item["id"].asString();	35 ✔
110	if (!Parameters.count(id)) {	35 ✔
111	continue;	32 ✔
112	}
113	auto deps = Expressions::GetDependencies(item["condition"].asString(), exprCache);	3 ✔
114	neededParams.insert(deps.begin(), deps.end());	3 ✔
115	}	35 ✔
116	if (neededParams.empty()) {	7 ✔
117	return {};	1 ✔
118	}
119	return CreateParametersRegisterList(ProtocolParams, Device, params, Json::Value(), neededParams);	6 ✔
120	}	7 ✔
121
122	TRPCRegisterList TRPCDeviceLoadRequest::GetChannelsRegisterList(const Json::Value& conditionParams)	8 ✔
123	{
124	auto notFound = Channels;	8 ✔
125	auto allChannels = Channels.empty();	8 ✔
126	Json::Value items(Json::arrayValue);	8 ✔
127	for (const auto& item: DeviceTemplate->GetTemplate()["channels"]) {	104 ✔
128	if (item["address"].isNull()) { // write only channel	48 ✔
129	continue;	8 ✔
130	}
131	auto id = item["name"].asString();	40 ✔
132	if (allChannels \|\| Channels.count(id)) {	40 ✔
133	auto channel = item;	35 ✔
134	channel["id"] = id;	35 ✔
135	items.append(channel);	35 ✔
136	notFound.erase(id);	35 ✔
137	}	35 ✔
138	}	40 ✔
139	if (!notFound.empty()) {	8 ✔
140	throw TRPCException("Channel \"" + *notFound.begin() + "\" is write only or not found in \"" +	×
141	DeviceTemplate->Type + "\" device template",	×
142	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
143	}
144
145	// Filter channels by condition using actual device parameter values,
146	// conditions over parameters without a value are evaluated with an undefined value
147	if (!conditionParams.isNull()) {	8 ✔
148	Expressions::TExpressionsCache cache;	7 ✔
149	TJsonParams exprParams(conditionParams);	7 ✔
150	Json::Value filtered(Json::arrayValue);	7 ✔
151	for (const auto& item: items) {	67 ✔
152	if (CheckCondition(item, exprParams, &cache)) {	30 ✔
153	filtered.append(item);	19 ✔
154	}
155	}
156	items = filtered;	7 ✔
157	}	7 ✔
158
159	return CreateChannelsRegisterList(ProtocolParams, Device, items);	16 ✔
160	}	8 ✔
161
162	TRPCRegisterList TRPCDeviceLoadRequest::GetParametersRegisterList(const Json::Value& conditionParams, Json::Value* data)	13 ✔
163	{
164	const auto& params = DeviceTemplate->GetTemplate()["parameters"];	13 ✔
165	auto notFound = Parameters;	13 ✔
166	Expressions::TExpressionsCache cache;	13 ✔
167	TJsonParams exprParams(conditionParams);	13 ✔
168	TActiveParameterDeclarations matched;	13 ✔
169	Json::Value items(Json::arrayValue);	13 ✔
170	for (auto it = params.begin(); it != params.end(); ++it) {	195 ✔
171	auto item = *it;	92 ✔
172	if (item["address"].isNull()) {	92 ✔
173	continue;	×
174	}
175	auto id = params.isObject() ? it.key().asString() : item["id"].asString();	92 ✔
176	if (!Parameters.count(id)) {	92 ✔
177	continue;	71 ✔
178	}
179	notFound.erase(id);	21 ✔
180	if (!conditionParams.isNull()) {	21 ✔
181	// conditions over parameters without a value are evaluated with an undefined value
182	if (!CheckCondition(item, exprParams, &cache)) {	21 ✔
183	continue;	8 ✔
184	}
185	// A chain of fw variants is not ambiguous, CreateParametersRegisterList merges it into one register.
186	// Anything else is a template error, config validation reports it as a duplicate definition
187	if (!matched.Add(id, *it)) {	13 ✔
188	throw TRPCException("Parameter \"" + id +	1 ✔
189	"\" is ambiguous: several declarations match the condition parameter values",	1 ✔
190	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	3 ✔
191	}
192	}
193	if (conditionParams.isMember(id)) {	12 ✔
194	// already read for condition evaluation, declarations of such a parameter read identically
195	if (data != nullptr) {	1 ✔
196	(*data)[id] = conditionParams[id];	1 ✔
197	}
198	continue;	1 ✔
199	}
200	if (params.isObject()) {	11 ✔
201	item["id"] = id;	×
202	}
203	items.append(item);	11 ✔
204	}	173 ✔
205	if (!notFound.empty()) {	12 ✔
206	throw TRPCException("Parameter \"" + *notFound.begin() + "\" is write only or not found in \"" +	2 ✔
207	DeviceTemplate->Type + "\" device template",	2 ✔
208	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	3 ✔
209	}
210
211	return CreateParametersRegisterList(ProtocolParams, Device, items);	22 ✔
212	}	21 ✔
213
214	PRPCDeviceLoadRequest ParseRPCDeviceLoadRequest(const Json::Value& request,	×
215	const TDeviceProtocolParams& protocolParams,
216	PSerialDevice device,
217	PDeviceTemplate deviceTemplate,
218	bool deviceFromConfig,
219	WBMQTT::TMqttRpcServer::TResultCallback onResult,
220	WBMQTT::TMqttRpcServer::TErrorCallback onError)
221	{
222	auto res = std::make_shared<TRPCDeviceLoadRequest>(protocolParams, device, deviceTemplate, deviceFromConfig);	×
223	res->ParseSettings(request, onResult, onError);	×
224	res->ParseRequestItems(request["channels"], res->Channels);	×
225	res->ParseRequestItems(request["parameters"], res->Parameters);	×
226	return res;	×
227	}	×
228
229	TRPCDeviceLoadSerialClientTask::TRPCDeviceLoadSerialClientTask(PRPCDeviceLoadRequest request): Request(request)	2 ✔
230	{
231	ExpireTime = std::chrono::steady_clock::now() + Request->TotalTimeout;	2 ✔
232	}	2 ✔
233
234	ISerialClientTask::TRunResult TRPCDeviceLoadSerialClientTask::Run(PFeaturePort port,	2 ✔
235	TSerialClientDeviceAccessHandler& lastAccessedDevice,
236	const std::list<PSerialDevice>& polledDevices)
237	{
238	if (IsCancelled()) {	2 ✔
NEW 239	AnswerRequestCancelled(Request->OnError);	×
NEW 240	return ISerialClientTask::TRunResult::OK;	×
241	}
242	if (std::chrono::steady_clock::now() > ExpireTime) {	2 ✔
243	if (Request->OnError) {	×
244	Request->OnError(WBMQTT::E_RPC_REQUEST_TIMEOUT, "RPC request timeout");	×
245	}
246	return ISerialClientTask::TRunResult::OK;	×
247	}
248	try {
249	if (!port->IsOpen()) {	2 ✔
250	port->Open();	2 ✔
251	}
252	lastAccessedDevice.PrepareToAccess(*port, nullptr);	2 ✔
253	if (!Request->DeviceFromConfig) {	2 ✔
254	TSerialPortSettingsGuard settingsGuard(port, Request->SerialPortSettings);	×
255	ExecRPCRequest(port, Request);	×
256	} else {	×
257	ExecRPCRequest(port, Request);	2 ✔
258	}
259	} catch (const std::exception& error) {	×
260	if (Request->OnError) {	×
261	Request->OnError(WBMQTT::E_RPC_SERVER_ERROR, std::string("Port IO error: ") + error.what());	×
262	}
263	}	×
264	return ISerialClientTask::TRunResult::OK;	2 ✔
265	}

wirenboard / wb-mqtt-serial / 967

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