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

0.0

/src/main.cpp

#include "log.h"
#include "serial_device.h"
#include "serial_driver.h"

#include <wblib/rpc.h>
#include <wblib/signal_handling.h>
#include <wblib/wbmqtt.h>

#include <filesystem>
#include <fstream>
#include <getopt.h>
#include <unistd.h>

#include "confed_json_generator.h"
#include "confed_schema_generator.h"
#include "config_schema_generator.h"

#include "device_template_generator.h"
#include "file_utils.h"
#include "files_watcher.h"
#include "port/serial_port.h"
#include "rpc/rpc_config_handler.h"
#include "rpc/rpc_device_handler.h"
#include "rpc/rpc_fw_update_handler.h"
#include "rpc/rpc_port_handler.h"
#include "rpc/rpc_templates_handler.h"

#define STR(x) #x
#define XSTR(x) STR(x)

using namespace std;

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

const auto driverName = "wb-modbus";

const auto APP_NAME = "wb-mqtt-serial";

const auto LIBWBMQTT_DB_FULL_FILE_PATH = "/var/lib/wb-mqtt-serial/libwbmqtt.db";
const auto CONFIG_FULL_FILE_PATH = "/etc/wb-mqtt-serial.conf";
const auto TEMPLATES_DIR = "/usr/share/wb-mqtt-serial/templates";
const auto USER_TEMPLATES_DIR = "/etc/wb-mqtt-serial.conf.d/templates";
const auto PORTS_JSON_SCHEMA_FULL_FILE_PATH = "/usr/share/wb-mqtt-serial/wb-mqtt-serial-ports.schema.json";
const auto TEMPLATES_JSON_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-device-template.schema.json";
const auto RPC_PORT_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-port-load-request.schema.json";
const auto RPC_PORT_SETUP_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-port-setup-request.schema.json";
const auto RPC_PORT_SCAN_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-port-scan-request.schema.json";
const auto RPC_DEVICE_LOAD_CONFIG_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-load-config-request.schema.json";
const auto RPC_DEVICE_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-load-request.schema.json";
const auto RPC_DEVICE_SET_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-set-request.schema.json";
const auto RPC_DEVICE_PROBE_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-probe-request.schema.json";
const auto RPC_DEVICE_SET_POLL_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-set-poll-request.schema.json";
const auto RPC_TEMPLATES_UPLOAD_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-templates-upload-request.schema.json";
const auto RPC_TEMPLATES_DELETE_REQUEST_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-templates-delete-request.schema.json";
const auto CONFED_COMMON_JSON_SCHEMA_FULL_FILE_PATH =
    "/usr/share/wb-mqtt-serial/wb-mqtt-serial-confed-common.schema.json";
const auto DEVICE_GROUP_NAMES_JSON_FULL_FILE_PATH = "/usr/share/wb-mqtt-serial/groups.json";
const auto PROTOCOL_SCHEMAS_DIR = "/usr/share/wb-mqtt-serial/protocols";

const auto SERIAL_DRIVER_STOP_TIMEOUT_S = chrono::seconds(60);

namespace
{
    void PrintStartupInfo()
    {
        string commit(XSTR(WBMQTT_COMMIT));
        cout << APP_NAME << " " << XSTR(WBMQTT_VERSION);
        if (!commit.empty()) {
            cout << " git " << commit;
        }
        cout << endl;
    }

    void PrintUsage()
    {
        cout << "Usage:" << endl
             << " " << APP_NAME << " [options]" << endl
             << "Options:" << endl
             << "  -d       level     enable debugging output:" << endl
             << "                       1 - serial only;" << endl
             << "                       2 - mqtt only;" << endl
             << "                       3 - both;" << endl
             << "                       negative values - silent mode (-1, -2, -3))" << endl
             << "  -c       config    config file" << endl
             << "  -p       port      MQTT broker port (default: 1883)" << endl
             << "  -h, -H   IP        MQTT broker IP (default: localhost)" << endl
             << "  -u       user      MQTT user (optional)" << endl
             << "  -P       password  MQTT user password (optional)" << endl
             << "  -T       prefix    MQTT topic prefix (optional)" << endl
             << "  -G       options   Generate device template. Type \"-G help\" for options description" << endl
             << "  -v                 Print the version" << endl;
    }

    pair<shared_ptr<Json::Value>, PTemplateMap> LoadTemplates()
    {
        auto commonDeviceSchema =
            make_shared<Json::Value>(WBMQTT::JSON::Parse(CONFED_COMMON_JSON_SCHEMA_FULL_FILE_PATH));
        auto templates = make_shared<TTemplateMap>(
            LoadConfigTemplatesSchema(TEMPLATES_JSON_SCHEMA_FULL_FILE_PATH, *commonDeviceSchema),
            USER_TEMPLATES_DIR);
        templates->AddTemplatesDir(TEMPLATES_DIR);
        // templates/Upload creates the directory
        try {
            templates->AddTemplatesDir(USER_TEMPLATES_DIR);
        } catch (const TNoDirError& e) {
            LOG(Warn) << "No user templates are loaded: " << e.what();
        }
        return {commonDeviceSchema, templates};
    }

    void SetDebugLevel(const char* optarg)
    {
        try {
            auto debugLevel = stoi(optarg);
            switch (debugLevel) {
                case 0:
                    return;
                case -1:
                    Info.SetEnabled(false);
                    return;

                case -2:
                    WBMQTT::Info.SetEnabled(false);
                    return;

                case -3:
                    WBMQTT::Info.SetEnabled(false);
                    Info.SetEnabled(false);
                    return;

                case 1:
                    Debug.SetEnabled(true);
                    return;

                case 2:
                    WBMQTT::Debug.SetEnabled(true);
                    return;

                case 3:
                    WBMQTT::Debug.SetEnabled(true);
                    Debug.SetEnabled(true);
                    return;
            }
        } catch (...) {
        }
        cout << "Invalid -d parameter value " << optarg << endl;
        PrintUsage();
        exit(2);
    }

    void ParseCommadLine(int argc, char* argv[], WBMQTT::TMosquittoMqttConfig& mqttConfig, string& customConfig)
    {
        int c;

        while ((c = getopt(argc, argv, "d:c:h:H:p:u:P:T:G:v")) != -1) {
            switch (c) {
                case 'd':
                    SetDebugLevel(optarg);
                    break;
                case 'c':
                    customConfig = optarg;
                    break;
                case 'p':
                    mqttConfig.Port = stoi(optarg);
                    break;
                case 'h':
                case 'H': // backward compatibility
                    mqttConfig.Host = optarg;
                    break;
                case 'T':
                    mqttConfig.Prefix = optarg;
                    break;
                case 'u':
                    mqttConfig.User = optarg;
                    break;
                case 'P':
                    mqttConfig.Password = optarg;
                    break;
                    exit(EXIT_SUCCESS);
                case 'G':
                    GenerateDeviceTemplate(APP_NAME, USER_TEMPLATES_DIR, optarg);
                    exit(EXIT_SUCCESS);
                case 'v':
                    PrintStartupInfo();
                    exit(EXIT_SUCCESS);
                case '?':
                default:
                    PrintStartupInfo();
                    PrintUsage();
                    exit(2);
            }
        }

        if (optind < argc) {
            for (int index = optind; index < argc; ++index) {
                cout << "Skipping unknown argument " << argv[index] << endl;
            }
        }
    }

    void HandleTemplateChangeEvent(TTemplateMap& templates,
                                   TDevicesConfedSchemasMap& confedSchemasMap,
                                   const std::string& fileName,
                                   TFilesWatcher::TEvent event)
    {
        if (event == TFilesWatcher::TEvent::CloseWrite) {
            LOG(Debug) << fileName << " changed. Reloading template";
            try {
                auto updatedTypes = templates.UpdateTemplate(fileName);
                for (const auto& deviceType: updatedTypes) {
                    confedSchemasMap.InvalidateCache(deviceType);
                }
            } catch (const exception& e) {
                LOG(Error) << "Failed to reload template: " << e.what();
            }
            return;
        }
        if (event == TFilesWatcher::TEvent::Delete) {
            LOG(Debug) << fileName << " deleted";
            confedSchemasMap.InvalidateCache(templates.DeleteTemplate(fileName));
        }
    }
}

int main(int argc, char* argv[])
{
    signal(SIGPIPE, SIG_IGN);

    WBMQTT::TMosquittoMqttConfig mqttConfig;
    string configFilename(CONFIG_FULL_FILE_PATH);

    WBMQTT::SignalHandling::Handle({SIGINT, SIGTERM});
    WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [&] { WBMQTT::SignalHandling::Stop(); });
    WBMQTT::SetThreadName(APP_NAME);

    ParseCommadLine(argc, argv, mqttConfig, configFilename);

    TSerialDeviceFactory deviceFactory;
    RegisterProtocols(deviceFactory);

    shared_ptr<Json::Value> commonDeviceSchema;
    PTemplateMap templates;
    std::tie(commonDeviceSchema, templates) = LoadTemplates();
    TDevicesConfedSchemasMap confedSchemasMap(*templates, deviceFactory, *commonDeviceSchema);
    TProtocolConfedSchemasMap protocolSchemasMap(PROTOCOL_SCHEMAS_DIR, *commonDeviceSchema);
    auto portsSchema = WBMQTT::JSON::Parse(PORTS_JSON_SCHEMA_FULL_FILE_PATH);

    TFilesWatcher watcher({TEMPLATES_DIR, USER_TEMPLATES_DIR}, [&](std::string fileName, TFilesWatcher::TEvent event) {
        HandleTemplateChangeEvent(*templates, confedSchemasMap, fileName, event);
    });

    try {
        if (mqttConfig.Id.empty())
            mqttConfig.Id = driverName;

        auto mqtt = WBMQTT::NewMosquittoMqttClient(mqttConfig);

        auto rpcServer(WBMQTT::NewMqttRpcServer(mqtt, APP_NAME));

        auto groupTranslations = WBMQTT::JSON::Parse(DEVICE_GROUP_NAMES_JSON_FULL_FILE_PATH);

        PHandlerConfig handlerConfig;

        try {
            handlerConfig = LoadConfig(configFilename,
                                       deviceFactory,
                                       *commonDeviceSchema,
                                       *templates,
                                       portsSchema,
                                       protocolSchemasMap);
        } catch (const exception& e) {
            LOG(Error) << e.what();
        }

        auto debugFromCommandLine = Debug.IsEnabled();
        if (handlerConfig && handlerConfig->Debug) {
            Debug.SetEnabled(true);
        }

        // The driver is needed without a configuration too: config/Save can set one later
        auto backend = WBMQTT::NewDriverBackend(mqtt);
        auto driver = WBMQTT::NewDriver(WBMQTT::TDriverArgs{}
                                            .SetId(driverName)
                                            .SetBackend(backend)
                                            .SetUseStorage(true)
                                            .SetReownUnknownDevices(true)
                                            .SetStoragePath(LIBWBMQTT_DB_FULL_FILE_PATH));

        driver->StartLoop();
        WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [=] {
            driver->StopLoop();
            driver->Close();
        });

        driver->WaitForReady();
        TConfigLoader configLoader(configFilename,
                                   deviceFactory,
                                   *commonDeviceSchema,
                                   *templates,
                                   portsSchema,
                                   protocolSchemasMap);
        TSerialDriverCore serialDriverCore(driver, handlerConfig, configLoader, debugFromCommandLine);
        handlerConfig.reset();
        TSerialClientTaskRunner serialClientTaskRunner(serialDriverCore);

        TRPCConfigHandler rpcConfigHandler(configFilename,
                                           portsSchema,
                                           templates,
                                           confedSchemasMap,
                                           protocolSchemasMap,
                                           groupTranslations,
                                           serialDriverCore,
                                           rpcServer);
        TRPCTemplatesHandler rpcTemplatesHandler(USER_TEMPLATES_DIR,
                                                 configFilename,
                                                 templates,
                                                 confedSchemasMap,
                                                 groupTranslations,
                                                 RPC_TEMPLATES_UPLOAD_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                 RPC_TEMPLATES_DELETE_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                 serialDriverCore);
        RegisterTemplatesRpcHandlers(rpcTemplatesHandler, rpcServer);

        auto rpcPortHandler = std::make_shared<TRPCPortHandler>(RPC_PORT_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                                RPC_PORT_SETUP_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                                RPC_PORT_SCAN_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                                serialDriverCore,
                                                                serialClientTaskRunner,
                                                                rpcServer);
        auto rpcDeviceHandler =
            std::make_shared<TRPCDeviceHandler>(configFilename,
                                                RPC_DEVICE_LOAD_CONFIG_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                RPC_DEVICE_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                RPC_DEVICE_SET_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                RPC_DEVICE_PROBE_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                RPC_DEVICE_SET_POLL_REQUEST_SCHEMA_FULL_FILE_PATH,
                                                deviceFactory,
                                                templates,
                                                serialClientTaskRunner,
                                                rpcServer);

        // Register separate RPC server for non-blocking firmware update tasks
        auto fwUpdateRpcServer(WBMQTT::NewMqttRpcServer(mqtt, APP_NAME));
        auto rpcFwUpdateHandler =
            std::make_shared<TRPCFwUpdateHandler>(serialClientTaskRunner, fwUpdateRpcServer, mqtt);

        serialDriverCore.Start();
        rpcServer->Start();

        WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [&] {
            rpcServer->Stop();
            fwUpdateRpcServer->Stop();
            serialDriverCore.Stop();
            rpcFwUpdateHandler->Stop();
        });
        WBMQTT::SignalHandling::SetOnTimeout(SERIAL_DRIVER_STOP_TIMEOUT_S, [&] {
            LOG(Error) << "Driver takes too long to stop. Exiting.";
            exit(EXIT_FAILURE);
        });
        WBMQTT::SignalHandling::Start();
        WBMQTT::SignalHandling::Wait();
    } catch (const exception& e) {
        LOG(Error) << "FATAL: " << e.what();
        return 1;
    }
    return 0;
}

1	#include "log.h"
2	#include "serial_device.h"
3	#include "serial_driver.h"
4
5	#include <wblib/rpc.h>
6	#include <wblib/signal_handling.h>
7	#include <wblib/wbmqtt.h>
8
9	#include <filesystem>
10	#include <fstream>
11	#include <getopt.h>
12	#include <unistd.h>
13
14	#include "confed_json_generator.h"
15	#include "confed_schema_generator.h"
16	#include "config_schema_generator.h"
17
18	#include "device_template_generator.h"
19	#include "file_utils.h"
20	#include "files_watcher.h"
21	#include "port/serial_port.h"
22	#include "rpc/rpc_config_handler.h"
23	#include "rpc/rpc_device_handler.h"
24	#include "rpc/rpc_fw_update_handler.h"
25	#include "rpc/rpc_port_handler.h"
26	#include "rpc/rpc_templates_handler.h"
27
28	#define STR(x) #x
29	#define XSTR(x) STR(x)
30
31	using namespace std;
32
33	#define LOG(logger) ::logger.Log() << "[serial] "
34
35	const auto driverName = "wb-modbus";
36
37	const auto APP_NAME = "wb-mqtt-serial";
38
39	const auto LIBWBMQTT_DB_FULL_FILE_PATH = "/var/lib/wb-mqtt-serial/libwbmqtt.db";
40	const auto CONFIG_FULL_FILE_PATH = "/etc/wb-mqtt-serial.conf";
41	const auto TEMPLATES_DIR = "/usr/share/wb-mqtt-serial/templates";
42	const auto USER_TEMPLATES_DIR = "/etc/wb-mqtt-serial.conf.d/templates";
43	const auto PORTS_JSON_SCHEMA_FULL_FILE_PATH = "/usr/share/wb-mqtt-serial/wb-mqtt-serial-ports.schema.json";
44	const auto TEMPLATES_JSON_SCHEMA_FULL_FILE_PATH =
45	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-device-template.schema.json";
46	const auto RPC_PORT_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH =
47	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-port-load-request.schema.json";
48	const auto RPC_PORT_SETUP_REQUEST_SCHEMA_FULL_FILE_PATH =
49	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-port-setup-request.schema.json";
50	const auto RPC_PORT_SCAN_REQUEST_SCHEMA_FULL_FILE_PATH =
51	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-port-scan-request.schema.json";
52	const auto RPC_DEVICE_LOAD_CONFIG_REQUEST_SCHEMA_FULL_FILE_PATH =
53	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-load-config-request.schema.json";
54	const auto RPC_DEVICE_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH =
55	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-load-request.schema.json";
56	const auto RPC_DEVICE_SET_REQUEST_SCHEMA_FULL_FILE_PATH =
57	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-set-request.schema.json";
58	const auto RPC_DEVICE_PROBE_REQUEST_SCHEMA_FULL_FILE_PATH =
59	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-probe-request.schema.json";
60	const auto RPC_DEVICE_SET_POLL_REQUEST_SCHEMA_FULL_FILE_PATH =
61	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-device-set-poll-request.schema.json";
62	const auto RPC_TEMPLATES_UPLOAD_REQUEST_SCHEMA_FULL_FILE_PATH =
63	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-templates-upload-request.schema.json";
64	const auto RPC_TEMPLATES_DELETE_REQUEST_SCHEMA_FULL_FILE_PATH =
65	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-rpc-templates-delete-request.schema.json";
66	const auto CONFED_COMMON_JSON_SCHEMA_FULL_FILE_PATH =
67	"/usr/share/wb-mqtt-serial/wb-mqtt-serial-confed-common.schema.json";
68	const auto DEVICE_GROUP_NAMES_JSON_FULL_FILE_PATH = "/usr/share/wb-mqtt-serial/groups.json";
69	const auto PROTOCOL_SCHEMAS_DIR = "/usr/share/wb-mqtt-serial/protocols";
70
71	const auto SERIAL_DRIVER_STOP_TIMEOUT_S = chrono::seconds(60);
72
73	namespace
74	{
75	void PrintStartupInfo()	×
76	{
77	string commit(XSTR(WBMQTT_COMMIT));	×
78	cout << APP_NAME << " " << XSTR(WBMQTT_VERSION);	×
79	if (!commit.empty()) {	×
80	cout << " git " << commit;	×
81	}
82	cout << endl;	×
83	}	×
84
85	void PrintUsage()	×
86	{
87	cout << "Usage:" << endl	×
88	<< " " << APP_NAME << " [options]" << endl	×
89	<< "Options:" << endl	×
90	<< " -d level enable debugging output:" << endl	×
91	<< " 1 - serial only;" << endl	×
92	<< " 2 - mqtt only;" << endl	×
93	<< " 3 - both;" << endl	×
94	<< " negative values - silent mode (-1, -2, -3))" << endl	×
95	<< " -c config config file" << endl	×
96	<< " -p port MQTT broker port (default: 1883)" << endl	×
97	<< " -h, -H IP MQTT broker IP (default: localhost)" << endl	×
98	<< " -u user MQTT user (optional)" << endl	×
99	<< " -P password MQTT user password (optional)" << endl	×
100	<< " -T prefix MQTT topic prefix (optional)" << endl	×
101	<< " -G options Generate device template. Type \"-G help\" for options description" << endl	×
102	<< " -v Print the version" << endl;	×
103	}	×
104
UNCOV 105	pair<shared_ptr<Json::Value>, PTemplateMap> LoadTemplates()	×
106	{
107	auto commonDeviceSchema =
108	make_shared<Json::Value>(WBMQTT::JSON::Parse(CONFED_COMMON_JSON_SCHEMA_FULL_FILE_PATH));	×
109	auto templates = make_shared<TTemplateMap>(
110	LoadConfigTemplatesSchema(TEMPLATES_JSON_SCHEMA_FULL_FILE_PATH, *commonDeviceSchema),	×
111	USER_TEMPLATES_DIR);	×
112	templates->AddTemplatesDir(TEMPLATES_DIR);	×
113	// templates/Upload creates the directory
114	try {
NEW 115	templates->AddTemplatesDir(USER_TEMPLATES_DIR);	×
NEW 116	} catch (const TNoDirError& e) {	×
NEW 117	LOG(Warn) << "No user templates are loaded: " << e.what();	×
118	}	×
NEW 119	return {commonDeviceSchema, templates};	×
UNCOV 120	}	×
121
122	void SetDebugLevel(const char* optarg)	×
123	{
124	try {
125	auto debugLevel = stoi(optarg);	×
126	switch (debugLevel) {	×
127	case 0:	×
128	return;	×
129	case -1:	×
130	Info.SetEnabled(false);	×
131	return;	×
132
133	case -2:	×
134	WBMQTT::Info.SetEnabled(false);	×
135	return;	×
136
137	case -3:	×
138	WBMQTT::Info.SetEnabled(false);	×
139	Info.SetEnabled(false);	×
140	return;	×
141
142	case 1:	×
143	Debug.SetEnabled(true);	×
144	return;	×
145
146	case 2:	×
147	WBMQTT::Debug.SetEnabled(true);	×
148	return;	×
149
150	case 3:	×
151	WBMQTT::Debug.SetEnabled(true);	×
152	Debug.SetEnabled(true);	×
153	return;	×
154	}
155	} catch (...) {	×
156	}	×
157	cout << "Invalid -d parameter value " << optarg << endl;	×
158	PrintUsage();	×
159	exit(2);	×
160	}
161
162	void ParseCommadLine(int argc, char* argv[], WBMQTT::TMosquittoMqttConfig& mqttConfig, string& customConfig)	×
163	{
164	int c;
165
NEW 166	while ((c = getopt(argc, argv, "d:c:h:H:p:u:P:T:G:v")) != -1) {	×
167	switch (c) {	×
168	case 'd':	×
169	SetDebugLevel(optarg);	×
170	break;	×
171	case 'c':	×
172	customConfig = optarg;	×
173	break;	×
174	case 'p':	×
175	mqttConfig.Port = stoi(optarg);	×
176	break;	×
177	case 'h':	×
178	case 'H': // backward compatibility
179	mqttConfig.Host = optarg;	×
180	break;	×
181	case 'T':	×
182	mqttConfig.Prefix = optarg;	×
183	break;	×
184	case 'u':	×
185	mqttConfig.User = optarg;	×
186	break;	×
187	case 'P':	×
188	mqttConfig.Password = optarg;	×
189	break;	×
190	exit(EXIT_SUCCESS);
191	case 'G':	×
192	GenerateDeviceTemplate(APP_NAME, USER_TEMPLATES_DIR, optarg);	×
193	exit(EXIT_SUCCESS);	×
194	case 'v':	×
195	PrintStartupInfo();	×
196	exit(EXIT_SUCCESS);	×
197	case '?':	×
198	default:
199	PrintStartupInfo();	×
200	PrintUsage();	×
201	exit(2);	×
202	}
203	}
204
205	if (optind < argc) {	×
206	for (int index = optind; index < argc; ++index) {	×
207	cout << "Skipping unknown argument " << argv[index] << endl;	×
208	}
209	}
210	}	×
211
212	void HandleTemplateChangeEvent(TTemplateMap& templates,	×
213	TDevicesConfedSchemasMap& confedSchemasMap,
214	const std::string& fileName,
215	TFilesWatcher::TEvent event)
216	{
217	if (event == TFilesWatcher::TEvent::CloseWrite) {	×
218	LOG(Debug) << fileName << " changed. Reloading template";	×
219	try {
220	auto updatedTypes = templates.UpdateTemplate(fileName);	×
221	for (const auto& deviceType: updatedTypes) {	×
222	confedSchemasMap.InvalidateCache(deviceType);	×
223	}
224	} catch (const exception& e) {	×
225	LOG(Error) << "Failed to reload template: " << e.what();	×
226	}	×
227	return;	×
228	}
229	if (event == TFilesWatcher::TEvent::Delete) {	×
230	LOG(Debug) << fileName << " deleted";	×
231	confedSchemasMap.InvalidateCache(templates.DeleteTemplate(fileName));	×
232	}
233	}
234	}
235
236	int main(int argc, char* argv[])	×
237	{
238	signal(SIGPIPE, SIG_IGN);	×
239
240	WBMQTT::TMosquittoMqttConfig mqttConfig;	×
241	string configFilename(CONFIG_FULL_FILE_PATH);	×
242
243	WBMQTT::SignalHandling::Handle({SIGINT, SIGTERM});	×
244	WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [&] { WBMQTT::SignalHandling::Stop(); });	×
245	WBMQTT::SetThreadName(APP_NAME);	×
246
247	ParseCommadLine(argc, argv, mqttConfig, configFilename);	×
248
249	TSerialDeviceFactory deviceFactory;	×
250	RegisterProtocols(deviceFactory);	×
251
252	shared_ptr<Json::Value> commonDeviceSchema;	×
253	PTemplateMap templates;	×
254	std::tie(commonDeviceSchema, templates) = LoadTemplates();	×
255	TDevicesConfedSchemasMap confedSchemasMap(templates, deviceFactory, commonDeviceSchema);	×
256	TProtocolConfedSchemasMap protocolSchemasMap(PROTOCOL_SCHEMAS_DIR, *commonDeviceSchema);	×
257	auto portsSchema = WBMQTT::JSON::Parse(PORTS_JSON_SCHEMA_FULL_FILE_PATH);	×
258
259	TFilesWatcher watcher({TEMPLATES_DIR, USER_TEMPLATES_DIR}, [&](std::string fileName, TFilesWatcher::TEvent event) {	×
260	HandleTemplateChangeEvent(*templates, confedSchemasMap, fileName, event);	×
261	});	×
262
263	try {
264	if (mqttConfig.Id.empty())	×
265	mqttConfig.Id = driverName;	×
266
267	auto mqtt = WBMQTT::NewMosquittoMqttClient(mqttConfig);	×
268
269	auto rpcServer(WBMQTT::NewMqttRpcServer(mqtt, APP_NAME));	×
270
271	auto groupTranslations = WBMQTT::JSON::Parse(DEVICE_GROUP_NAMES_JSON_FULL_FILE_PATH);	×
272
UNCOV 273	PHandlerConfig handlerConfig;	×
274
275	try {
276	handlerConfig = LoadConfig(configFilename,	×
277	deviceFactory,
278	*commonDeviceSchema,	×
279	*templates,	×
280	portsSchema,
281	protocolSchemasMap);	×
282	} catch (const exception& e) {	×
283	LOG(Error) << e.what();	×
284	}	×
285
NEW 286	auto debugFromCommandLine = Debug.IsEnabled();	×
NEW 287	if (handlerConfig && handlerConfig->Debug) {	×
NEW 288	Debug.SetEnabled(true);	×
289	}
290
291	// The driver is needed without a configuration too: config/Save can set one later
NEW 292	auto backend = WBMQTT::NewDriverBackend(mqtt);	×
NEW 293	auto driver = WBMQTT::NewDriver(WBMQTT::TDriverArgs{}	×
NEW 294	.SetId(driverName)	×
NEW 295	.SetBackend(backend)	×
NEW 296	.SetUseStorage(true)	×
NEW 297	.SetReownUnknownDevices(true)	×
NEW 298	.SetStoragePath(LIBWBMQTT_DB_FULL_FILE_PATH));	×
299
NEW 300	driver->StartLoop();	×
NEW 301	WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [=] {	×
NEW 302	driver->StopLoop();	×
NEW 303	driver->Close();	×
NEW 304	});	×
305
NEW 306	driver->WaitForReady();	×
307	TConfigLoader configLoader(configFilename,
308	deviceFactory,
NEW 309	*commonDeviceSchema,	×
NEW 310	*templates,	×
311	portsSchema,
NEW 312	protocolSchemasMap);	×
NEW 313	TSerialDriverCore serialDriverCore(driver, handlerConfig, configLoader, debugFromCommandLine);	×
NEW 314	handlerConfig.reset();	×
NEW 315	TSerialClientTaskRunner serialClientTaskRunner(serialDriverCore);	×
316
317	TRPCConfigHandler rpcConfigHandler(configFilename,
318	portsSchema,
319	templates,
320	confedSchemasMap,
321	protocolSchemasMap,
322	groupTranslations,
323	serialDriverCore,
NEW 324	rpcServer);	×
325	TRPCTemplatesHandler rpcTemplatesHandler(USER_TEMPLATES_DIR,
326	configFilename,
327	templates,
328	confedSchemasMap,
329	groupTranslations,
330	RPC_TEMPLATES_UPLOAD_REQUEST_SCHEMA_FULL_FILE_PATH,
331	RPC_TEMPLATES_DELETE_REQUEST_SCHEMA_FULL_FILE_PATH,
NEW 332	serialDriverCore);	×
NEW 333	RegisterTemplatesRpcHandlers(rpcTemplatesHandler, rpcServer);	×
334
335	auto rpcPortHandler = std::make_shared<TRPCPortHandler>(RPC_PORT_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH,
336	RPC_PORT_SETUP_REQUEST_SCHEMA_FULL_FILE_PATH,
337	RPC_PORT_SCAN_REQUEST_SCHEMA_FULL_FILE_PATH,
338	serialDriverCore,
339	serialClientTaskRunner,
UNCOV 340	rpcServer);	×
341	auto rpcDeviceHandler =
342	std::make_shared<TRPCDeviceHandler>(configFilename,
343	RPC_DEVICE_LOAD_CONFIG_REQUEST_SCHEMA_FULL_FILE_PATH,
344	RPC_DEVICE_LOAD_REQUEST_SCHEMA_FULL_FILE_PATH,
345	RPC_DEVICE_SET_REQUEST_SCHEMA_FULL_FILE_PATH,
346	RPC_DEVICE_PROBE_REQUEST_SCHEMA_FULL_FILE_PATH,
347	RPC_DEVICE_SET_POLL_REQUEST_SCHEMA_FULL_FILE_PATH,
348	deviceFactory,
349	templates,
350	serialClientTaskRunner,
UNCOV 351	rpcServer);	×
352
353	// Register separate RPC server for non-blocking firmware update tasks
354	auto fwUpdateRpcServer(WBMQTT::NewMqttRpcServer(mqtt, APP_NAME));	×
355	auto rpcFwUpdateHandler =
356	std::make_shared<TRPCFwUpdateHandler>(serialClientTaskRunner, fwUpdateRpcServer, mqtt);	×
357
NEW 358	serialDriverCore.Start();	×
359	rpcServer->Start();	×
360
NEW 361	WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [&] {	×
362	rpcServer->Stop();	×
363	fwUpdateRpcServer->Stop();	×
NEW 364	serialDriverCore.Stop();	×
NEW 365	rpcFwUpdateHandler->Stop();	×
366	});	×
367	WBMQTT::SignalHandling::SetOnTimeout(SERIAL_DRIVER_STOP_TIMEOUT_S, [&] {	×
368	LOG(Error) << "Driver takes too long to stop. Exiting.";	×
369	exit(EXIT_FAILURE);	×
370	});
371	WBMQTT::SignalHandling::Start();	×
372	WBMQTT::SignalHandling::Wait();	×
373	} catch (const exception& e) {	×
374	LOG(Error) << "FATAL: " << e.what();	×
375	return 1;	×
376	}	×
377	return 0;	×
378	}	×

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