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

84.85

/src/serial_client_task_executor.cpp

#include "serial_client_task_executor.h"
#include "log.h"

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

TSerialClientTaskExecutor::TSerialClientTaskExecutor(PFeaturePort port)
    : Port(port),
      StopRequested(false),
      Running(true),
      Idle(true)
{
    Thread = std::thread([this]() {
        TSerialClientDeviceAccessHandler lastAccessedDevice(nullptr);
        while (true) {
            std::unique_lock<std::mutex> lock(Mutex);
            TasksCv.wait(lock, [this]() { return !Tasks.empty() || !Running; });
            if (!Running) {
                break;
            }
            if (Tasks.empty()) {
                continue;
            }
            Idle = false;
            // One by one, so RequestStop cancels every task which has not started
            while (!Tasks.empty()) {
                auto task = Tasks.front();
                Tasks.pop_front();
                lock.unlock();
                try {
                    task->Run(Port, lastAccessedDevice, std::list<PSerialDevice>());
                } catch (const std::exception& e) {
                    LOG(Error) << "Error while running task: " << e.what();
                }
                lock.lock();
            }
            Idle = true;
            Port->Close();
            IdleCv.notify_all();
        }
    });
}

TSerialClientTaskExecutor::~TSerialClientTaskExecutor()
{
    {
        std::unique_lock<std::mutex> lock(Mutex);
        Running = false;
    }
    TasksCv.notify_all();
    Thread.join();
}

void TSerialClientTaskExecutor::AddTask(PSerialClientTask task)
{
    if (!Running) {
        return;
    }
    {
        std::unique_lock<std::mutex> lock(Mutex);
        if (StopRequested) {
            task->Cancel();
        }
        Tasks.push_back(task);
    }
    TasksCv.notify_all();
}

void TSerialClientTaskExecutor::RequestStop()
{
    std::unique_lock<std::mutex> lock(Mutex);
    StopRequested = true;
    for (const auto& task: Tasks) {
        task->Cancel();
    }
}

bool TSerialClientTaskExecutor::WaitStopped(std::chrono::steady_clock::time_point deadline)
{
    std::unique_lock<std::mutex> lock(Mutex);
    return IdleCv.wait_until(lock, deadline, [this]() { return Idle; });
}

void TSerialClientTaskExecutor::Resume()
{
    std::unique_lock<std::mutex> lock(Mutex);
    StopRequested = false;
}

PFeaturePort TSerialClientTaskExecutor::GetPort() const
{
    return Port;
}

bool TSerialClientTaskExecutor::IsIdle() const
{
    std::unique_lock<std::mutex> lock(Mutex);
    return Idle;
}

1	#include "serial_client_task_executor.h"
2	#include "log.h"
3
4	#define LOG(logger) ::logger.Log() << "[serial] "
5
6	TSerialClientTaskExecutor::TSerialClientTaskExecutor(PFeaturePort port)	7 ✔
7	: Port(port),	7 ✔
8	StopRequested(false),	7 ✔
9	Running(true),	7 ✔
10	Idle(true)	7 ✔
11	{
12	Thread = std::thread([this]() {	7 ✔
13	TSerialClientDeviceAccessHandler lastAccessedDevice(nullptr);	7 ✔
14	while (true) {
15	std::unique_lock<std::mutex> lock(Mutex);	16 ✔
16	TasksCv.wait(lock, [this]() { return !Tasks.empty() \|\| !Running; });	44 ✔
17	if (!Running) {	16 ✔
18	break;	7 ✔
19	}
20	if (Tasks.empty()) {	9 ✔
NEW 21	continue;	×
22	}
23	Idle = false;	9 ✔
24	// One by one, so RequestStop cancels every task which has not started
25	while (!Tasks.empty()) {	20 ✔
26	auto task = Tasks.front();	11 ✔
27	Tasks.pop_front();	11 ✔
28	lock.unlock();	11 ✔
29	try {
30	task->Run(Port, lastAccessedDevice, std::list<PSerialDevice>());	11 ✔
NEW 31	} catch (const std::exception& e) {	×
NEW 32	LOG(Error) << "Error while running task: " << e.what();	×
NEW 33	}	×
34	lock.lock();	11 ✔
35	}	11 ✔
36	Idle = true;	9 ✔
37	Port->Close();	9 ✔
38	IdleCv.notify_all();	9 ✔
39	}	25 ✔
40	});	14 ✔
41	}	7 ✔
42
43	TSerialClientTaskExecutor::~TSerialClientTaskExecutor()	7 ✔
44	{
45	{
46	std::unique_lock<std::mutex> lock(Mutex);	7 ✔
47	Running = false;	7 ✔
48	}	7 ✔
49	TasksCv.notify_all();	7 ✔
50	Thread.join();	7 ✔
51	}	7 ✔
52
53	void TSerialClientTaskExecutor::AddTask(PSerialClientTask task)	11 ✔
54	{
55	if (!Running) {	11 ✔
NEW 56	return;	×
57	}
58	{
59	std::unique_lock<std::mutex> lock(Mutex);	11 ✔
60	if (StopRequested) {	11 ✔
NEW 61	task->Cancel();	×
62	}
63	Tasks.push_back(task);	11 ✔
64	}	11 ✔
65	TasksCv.notify_all();	11 ✔
66	}
67
68	void TSerialClientTaskExecutor::RequestStop()	8 ✔
69	{
70	std::unique_lock<std::mutex> lock(Mutex);	8 ✔
71	StopRequested = true;	8 ✔
72	for (const auto& task: Tasks) {	10 ✔
73	task->Cancel();	2 ✔
74	}
75	}	8 ✔
76
77	bool TSerialClientTaskExecutor::WaitStopped(std::chrono::steady_clock::time_point deadline)	5 ✔
78	{
79	std::unique_lock<std::mutex> lock(Mutex);	5 ✔
80	return IdleCv.wait_until(lock, deadline, [this]() { return Idle; });	17 ✔
81	}	5 ✔
82
83	void TSerialClientTaskExecutor::Resume()	2 ✔
84	{
85	std::unique_lock<std::mutex> lock(Mutex);	2 ✔
86	StopRequested = false;	2 ✔
87	}	2 ✔
88
89	PFeaturePort TSerialClientTaskExecutor::GetPort() const	4 ✔
90	{
91	return Port;	4 ✔
92	}
93
NEW 94	bool TSerialClientTaskExecutor::IsIdle() const	×
95	{
NEW 96	std::unique_lock<std::mutex> lock(Mutex);	×
NEW 97	return Idle;	×
NEW 98	}	×

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