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

81.94

/src/rpc/rpc_device_set_task.cpp

#include "rpc_device_set_task.h"
#include "config_merge_template.h"
#include "expression_evaluator.h"
#include "rpc_helpers.h"

#include <set>

#define LOG(logger) ::logger.Log() << "[RPC] "

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

        // 1. Template declarations of the requested channels and parameters, every condition variant
        auto channelDeclarations = rpcRequest->GetChannelDeclarations();
        auto parameterDeclarations = rpcRequest->GetParameterDeclarations();
        if (channelDeclarations.empty() && parameterDeclarations.empty()) {
            rpcRequest->OnResult(Json::Value(Json::objectValue));
            return;
        }

        PrepareSession(*port, rpcRequest->Device);

        std::string error;
        try {
            // 2. The request parameters are the condition inputs, the missing condition parameters are read
            Json::Value conditionValues(rpcRequest->Parameters);
            auto registerList =
                rpcRequest->GetConditionParametersRegisterList(channelDeclarations, parameterDeclarations);
            ReadRegisterList(*port, rpcRequest->Device, registerList);
            MarkUnsupportedRegisterItems(*port, *rpcRequest, registerList);
            MergeRegisterListValues(registerList, conditionValues);

            // 3. The declaration with a true condition acts, an item without one is skipped
            auto channels = rpcRequest->SelectActingDeclarations("Channel", channelDeclarations, conditionValues);
            auto parameters = rpcRequest->SelectActingDeclarations("Parameter", parameterDeclarations, conditionValues);

            // 4. Write
            rpcRequest->Device->WriteSetupRegisters(*port, rpcRequest->CreateSetupItems(channels, parameters), true);
        } catch (const TSerialDeviceException& e) {
            error = e.what();
        } catch (...) {
            EndSession(*port, rpcRequest->Device);
            throw;
        }

        EndSession(*port, rpcRequest->Device);

        if (!error.empty()) {
            LOG(Warn) << port->GetDescription() << rpcRequest->Device->ToString() << ": " << error;
            throw TRPCException(error, TRPCResultCode::RPC_WRONG_PARAM_VALUE);
        }

        rpcRequest->OnResult(Json::Value(Json::objectValue));
    }
} // namespace

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

Json::Value TRPCDeviceSetRequest::GetChannelDeclarations()
{
    Json::Value notFound(Channels);
    Json::Value declarations(Json::arrayValue);
    for (const auto& item: DeviceTemplate->GetTemplate()["channels"]) {
        auto name = item["name"].asString();
        if (!Channels.isMember(name) || IsReadOnly(item)) {
            continue;
        }
        notFound.removeMember(name);
        auto declaration = item;
        declaration["id"] = name;
        declarations.append(declaration);
    }
    if (!notFound.empty()) {
        throw TRPCException("Channel \"" + notFound.getMemberNames().front() + "\" is read only or not found in \"" +
                                DeviceTemplate->Type + "\" device template",
                            TRPCResultCode::RPC_WRONG_PARAM_VALUE);
    }
    return declarations;
}

Json::Value TRPCDeviceSetRequest::GetParameterDeclarations()
{
    const auto& params = DeviceTemplate->GetTemplate()["parameters"];
    Json::Value declarations(Json::arrayValue);
    for (auto it = params.begin(); it != params.end(); ++it) {
        auto id = params.isObject() ? it.key().asString() : (*it)["id"].asString();
        // read only declarations are never written, so their conditions are not evaluated
        if ((*it)["readonly"].asBool() || !Parameters.isMember(id)) {
            continue;
        }
        auto item = *it;
        item["id"] = id;
        declarations.append(item);
    }
    return declarations;
}

TRPCRegisterList TRPCDeviceSetRequest::GetConditionParametersRegisterList(const Json::Value& channels,
                                                                          const Json::Value& parameters)
{
    std::set<std::string> neededParams;
    Expressions::TExpressionsCache exprCache;
    for (const auto* declarations: {&channels, &parameters}) {
        for (const auto& item: *declarations) {
            auto deps = Expressions::GetDependencies(item["condition"].asString(), exprCache);
            neededParams.insert(deps.begin(), deps.end());
        }
    }
    for (const auto& name: Parameters.getMemberNames()) {
        neededParams.erase(name);
    }
    if (neededParams.empty()) {
        return {};
    }
    return CreateParametersRegisterList(ProtocolParams,
                                        Device,
                                        DeviceTemplate->GetTemplate()["parameters"],
                                        Json::Value(),
                                        neededParams);
}

Json::Value TRPCDeviceSetRequest::SelectActingDeclarations(const std::string& kind,
                                                           const Json::Value& declarations,
                                                           const Json::Value& conditionValues)
{
    TJsonParams exprParams(conditionValues);
    Expressions::TExpressionsCache exprCache;
    TActiveParameterDeclarations matched;
    Json::Value acting(Json::arrayValue);
    for (const auto& item: declarations) {
        if (!CheckCondition(item, exprParams, &exprCache)) {
            continue;
        }
        auto id = item["id"].asString();
        auto first = matched.GetNewest(id) == nullptr;
        // Several matched declarations of an item are allowed only as a chain of fw variants,
        // the chain acts as its first declaration since the variants define the same register.
        // Anything else is a template error, config validation reports it as a duplicate definition
        if (!matched.Add(id, item)) {
            throw TRPCException(kind + " \"" + id +
                                    "\" is ambiguous: several declarations match the condition parameter values",
                                TRPCResultCode::RPC_WRONG_PARAM_VALUE);
        }
        if (first) {
            acting.append(item);
        }
    }
    return acting;
}

TDeviceSetupItems TRPCDeviceSetRequest::CreateSetupItems(const Json::Value& channels, const Json::Value& parameters)
{
    TDeviceSetupItems setupItems;
    for (const auto& item: channels) {
        setupItems.insert(CreateSetupItem(item, Channels[item["id"].asString()].asString()));
    }
    for (const auto& item: parameters) {
        setupItems.insert(CreateSetupItem(item, Parameters[item["id"].asString()].asString()));
    }
    return setupItems;
}

PRegisterConfig TRPCDeviceSetRequest::GetRegisterConfig(const Json::Value& declaration)
{
    return LoadRegisterConfig(declaration,
                              *ProtocolParams.protocol->GetRegTypes(),
                              std::string(),
                              *ProtocolParams.factory,
                              ProtocolParams.factory->GetRegisterAddressFactory().GetBaseRegisterAddress(),
                              0)
        .RegisterConfig;
}

bool TRPCDeviceSetRequest::IsReadOnly(const Json::Value& declaration)
{
    return GetRegisterConfig(declaration)->AccessType == TRegisterConfig::EAccessType::READ_ONLY;
}

PDeviceSetupItem TRPCDeviceSetRequest::CreateSetupItem(const Json::Value& declaration, const std::string& value)
{
    auto itemConfig =
        std::make_shared<TDeviceSetupItemConfig>(declaration["id"].asString(), GetRegisterConfig(declaration), value);
    return std::make_shared<TDeviceSetupItem>(itemConfig, Device);
}

PRPCDeviceSetRequest ParseRPCDeviceSetRequest(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<TRPCDeviceSetRequest>(protocolParams, device, deviceTemplate, deviceFromConfig);
    res->ParseSettings(request, onResult, onError);
    res->Channels = request["channels"];
    res->Parameters = request["parameters"];
    return res;
}

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

ISerialClientTask::TRunResult TRPCDeviceSetSerialClientTask::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_set_task.h"
2	#include "config_merge_template.h"
3	#include "expression_evaluator.h"
4	#include "rpc_helpers.h"
5
6	#include <set>
7
8	#define LOG(logger) ::logger.Log() << "[RPC] "
9
10	namespace
11	{
12	void ExecRPCRequest(PPort port, PRPCDeviceSetRequest rpcRequest)	3 ✔
13	{
14	if (!rpcRequest->OnResult) {	3 ✔
15	return;	1 ✔
16	}
17
18	// 1. Template declarations of the requested channels and parameters, every condition variant
19	auto channelDeclarations = rpcRequest->GetChannelDeclarations();	3 ✔
20	auto parameterDeclarations = rpcRequest->GetParameterDeclarations();	3 ✔
21	if (channelDeclarations.empty() && parameterDeclarations.empty()) {	3 ✔
22	rpcRequest->OnResult(Json::Value(Json::objectValue));	1 ✔
23	return;	1 ✔
24	}
25
26	PrepareSession(*port, rpcRequest->Device);	2 ✔
27
28	std::string error;	2 ✔
29	try {
30	// 2. The request parameters are the condition inputs, the missing condition parameters are read
31	Json::Value conditionValues(rpcRequest->Parameters);	2 ✔
32	auto registerList =
33	rpcRequest->GetConditionParametersRegisterList(channelDeclarations, parameterDeclarations);	2 ✔
34	ReadRegisterList(*port, rpcRequest->Device, registerList);	2 ✔
35	MarkUnsupportedRegisterItems(port, rpcRequest, registerList);	2 ✔
36	MergeRegisterListValues(registerList, conditionValues);	2 ✔
37
38	// 3. The declaration with a true condition acts, an item without one is skipped
39	auto channels = rpcRequest->SelectActingDeclarations("Channel", channelDeclarations, conditionValues);	4 ✔
40	auto parameters = rpcRequest->SelectActingDeclarations("Parameter", parameterDeclarations, conditionValues);	4 ✔
41
42	// 4. Write
43	rpcRequest->Device->WriteSetupRegisters(*port, rpcRequest->CreateSetupItems(channels, parameters), true);	2 ✔
44	} catch (const TSerialDeviceException& e) {	2 ✔
45	error = e.what();	×
46	} catch (...) {	×
47	EndSession(*port, rpcRequest->Device);	×
48	throw;	×
49	}	×
50
51	EndSession(*port, rpcRequest->Device);	2 ✔
52
53	if (!error.empty()) {	2 ✔
54	LOG(Warn) << port->GetDescription() << rpcRequest->Device->ToString() << ": " << error;	×
55	throw TRPCException(error, TRPCResultCode::RPC_WRONG_PARAM_VALUE);	×
56	}
57
58	rpcRequest->OnResult(Json::Value(Json::objectValue));	2 ✔
59	}	4 ✔
60	} // namespace
61
62	TRPCDeviceSetRequest::TRPCDeviceSetRequest(const TDeviceProtocolParams& protocolParams,	24 ✔
63	PSerialDevice device,
64	PDeviceTemplate deviceTemplate,
65	bool deviceFromConfig)	24 ✔
66	: TRPCDeviceRequest(protocolParams, device, deviceTemplate, deviceFromConfig)	24 ✔
67	{}	24 ✔
68
69	Json::Value TRPCDeviceSetRequest::GetChannelDeclarations()	14 ✔
70	{
71	Json::Value notFound(Channels);	14 ✔
72	Json::Value declarations(Json::arrayValue);	14 ✔
73	for (const auto& item: DeviceTemplate->GetTemplate()["channels"]) {	154 ✔
74	auto name = item["name"].asString();	70 ✔
75	if (!Channels.isMember(name) \|\| IsReadOnly(item)) {	70 ✔
76	continue;	58 ✔
77	}
78	notFound.removeMember(name);	12 ✔
79	auto declaration = item;	12 ✔
80	declaration["id"] = name;	12 ✔
81	declarations.append(declaration);	12 ✔
82	}	70 ✔
83	if (!notFound.empty()) {	14 ✔
84	throw TRPCException("Channel \"" + notFound.getMemberNames().front() + "\" is read only or not found in \"" +	4 ✔
85	DeviceTemplate->Type + "\" device template",	4 ✔
86	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	6 ✔
87	}
88	return declarations;	12 ✔
89	}	16 ✔
90
91	Json::Value TRPCDeviceSetRequest::GetParameterDeclarations()	18 ✔
92	{
93	const auto& params = DeviceTemplate->GetTemplate()["parameters"];	18 ✔
94	Json::Value declarations(Json::arrayValue);	18 ✔
95	for (auto it = params.begin(); it != params.end(); ++it) {	356 ✔
96	auto id = params.isObject() ? it.key().asString() : (*it)["id"].asString();	169 ✔
97	// read only declarations are never written, so their conditions are not evaluated
98	if ((*it)["readonly"].asBool() \|\| !Parameters.isMember(id)) {	169 ✔
99	continue;	138 ✔
100	}
101	auto item = *it;	31 ✔
102	item["id"] = id;	31 ✔
103	declarations.append(item);	31 ✔
104	}	169 ✔
105	return declarations;	18 ✔
106	}	×
107
108	TRPCRegisterList TRPCDeviceSetRequest::GetConditionParametersRegisterList(const Json::Value& channels,	7 ✔
109	const Json::Value& parameters)
110	{
111	std::set<std::string> neededParams;	7 ✔
112	Expressions::TExpressionsCache exprCache;	7 ✔
113	for (const auto* declarations: {&channels, &parameters}) {	21 ✔
114	for (const auto& item: *declarations) {	46 ✔
115	auto deps = Expressions::GetDependencies(item["condition"].asString(), exprCache);	16 ✔
116	neededParams.insert(deps.begin(), deps.end());	16 ✔
117	}	16 ✔
118	}
119	for (const auto& name: Parameters.getMemberNames()) {	15 ✔
120	neededParams.erase(name);	8 ✔
121	}	7 ✔
122	if (neededParams.empty()) {	7 ✔
123	return {};	3 ✔
124	}
125	return CreateParametersRegisterList(ProtocolParams,	4 ✔
126	Device,	4 ✔
127	DeviceTemplate->GetTemplate()["parameters"],	4 ✔
128	Json::Value(),	4 ✔
129	neededParams);	4 ✔
130	}	7 ✔
131
132	Json::Value TRPCDeviceSetRequest::SelectActingDeclarations(const std::string& kind,	18 ✔
133	const Json::Value& declarations,
134	const Json::Value& conditionValues)
135	{
136	TJsonParams exprParams(conditionValues);	18 ✔
137	Expressions::TExpressionsCache exprCache;	18 ✔
138	TActiveParameterDeclarations matched;	18 ✔
139	Json::Value acting(Json::arrayValue);	18 ✔
140	for (const auto& item: declarations) {	84 ✔
141	if (!CheckCondition(item, exprParams, &exprCache)) {	35 ✔
142	continue;	17 ✔
143	}
144	auto id = item["id"].asString();	18 ✔
145	auto first = matched.GetNewest(id) == nullptr;	18 ✔
146	// Several matched declarations of an item are allowed only as a chain of fw variants,
147	// the chain acts as its first declaration since the variants define the same register.
148	// Anything else is a template error, config validation reports it as a duplicate definition
149	if (!matched.Add(id, item)) {	18 ✔
150	throw TRPCException(kind + " \"" + id +	4 ✔
151	"\" is ambiguous: several declarations match the condition parameter values",	2 ✔
152	TRPCResultCode::RPC_WRONG_PARAM_VALUE);	6 ✔
153	}
154	if (first) {	16 ✔
155	acting.append(item);	14 ✔
156	}
157	}	18 ✔
158	return acting;	16 ✔
159	}	24 ✔
160
161	TDeviceSetupItems TRPCDeviceSetRequest::CreateSetupItems(const Json::Value& channels, const Json::Value& parameters)	14 ✔
162	{
163	TDeviceSetupItems setupItems;	14 ✔
164	for (const auto& item: channels) {	17 ✔
165	setupItems.insert(CreateSetupItem(item, Channels[item["id"].asString()].asString()));	3 ✔
166	}
167	for (const auto& item: parameters) {	23 ✔
168	setupItems.insert(CreateSetupItem(item, Parameters[item["id"].asString()].asString()));	9 ✔
169	}
170	return setupItems;	14 ✔
171	}	×
172
173	PRegisterConfig TRPCDeviceSetRequest::GetRegisterConfig(const Json::Value& declaration)	25 ✔
174	{
175	return LoadRegisterConfig(declaration,	50 ✔
176	*ProtocolParams.protocol->GetRegTypes(),	50 ✔
177	std::string(),	50 ✔
178	*ProtocolParams.factory,	25 ✔
179	ProtocolParams.factory->GetRegisterAddressFactory().GetBaseRegisterAddress(),	25 ✔
180	0)
181	.RegisterConfig;	50 ✔
182	}
183
184	bool TRPCDeviceSetRequest::IsReadOnly(const Json::Value& declaration)	13 ✔
185	{
186	return GetRegisterConfig(declaration)->AccessType == TRegisterConfig::EAccessType::READ_ONLY;	13 ✔
187	}
188
189	PDeviceSetupItem TRPCDeviceSetRequest::CreateSetupItem(const Json::Value& declaration, const std::string& value)	12 ✔
190	{
191	auto itemConfig =
192	std::make_shared<TDeviceSetupItemConfig>(declaration["id"].asString(), GetRegisterConfig(declaration), value);	12 ✔
193	return std::make_shared<TDeviceSetupItem>(itemConfig, Device);	24 ✔
194	}	12 ✔
195
196	PRPCDeviceSetRequest ParseRPCDeviceSetRequest(const Json::Value& request,	×
197	const TDeviceProtocolParams& protocolParams,
198	PSerialDevice device,
199	PDeviceTemplate deviceTemplate,
200	bool deviceFromConfig,
201	WBMQTT::TMqttRpcServer::TResultCallback onResult,
202	WBMQTT::TMqttRpcServer::TErrorCallback onError)
203	{
204	auto res = std::make_shared<TRPCDeviceSetRequest>(protocolParams, device, deviceTemplate, deviceFromConfig);	×
205	res->ParseSettings(request, onResult, onError);	×
206	res->Channels = request["channels"];	×
207	res->Parameters = request["parameters"];	×
208	return res;	×
209	}	×
210
211	TRPCDeviceSetSerialClientTask::TRPCDeviceSetSerialClientTask(PRPCDeviceSetRequest request): Request(request)	3 ✔
212	{
213	ExpireTime = std::chrono::steady_clock::now() + Request->TotalTimeout;	3 ✔
214	}	3 ✔
215
216	ISerialClientTask::TRunResult TRPCDeviceSetSerialClientTask::Run(PFeaturePort port,	3 ✔
217	TSerialClientDeviceAccessHandler& lastAccessedDevice,
218	const std::list<PSerialDevice>& polledDevices)
219	{
220	if (IsCancelled()) {	3 ✔
NEW 221	AnswerRequestCancelled(Request->OnError);	×
NEW 222	return ISerialClientTask::TRunResult::OK;	×
223	}
224	if (std::chrono::steady_clock::now() > ExpireTime) {	3 ✔
225	if (Request->OnError) {	×
226	Request->OnError(WBMQTT::E_RPC_REQUEST_TIMEOUT, "RPC request timeout");	×
227	}
228	return ISerialClientTask::TRunResult::OK;	×
229	}
230	try {
231	if (!port->IsOpen()) {	3 ✔
232	port->Open();	3 ✔
233	}
234	lastAccessedDevice.PrepareToAccess(*port, nullptr);	3 ✔
235	if (!Request->DeviceFromConfig) {	3 ✔
236	TSerialPortSettingsGuard settingsGuard(port, Request->SerialPortSettings);	×
237	ExecRPCRequest(port, Request);	×
238	} else {	×
239	ExecRPCRequest(port, Request);	3 ✔
240	}
241	} catch (const std::exception& error) {	×
242	if (Request->OnError) {	×
243	Request->OnError(WBMQTT::E_RPC_SERVER_ERROR, std::string("Port IO error: ") + error.what());	×
244	}
245	}	×
246	return ISerialClientTask::TRunResult::OK;	3 ✔
247	}

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