wirenboard / wb-mqtt-db / 96

Committed 31 Jul 2026 10:43AM UTC coverage: 78.543% (-1.8%) from 80.337%

Build # 96

Build Type

push

github

Committed by

web-flow

Commit Message

Use exported vars instead of MAKEFLAGS for coverage options (#66)

Coverage Stats

938 of 1102 branches covered (85.12%)

1563 of 1990 relevant lines covered (78.54%)

13.11 hits per line

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

0.0

/src/main.cpp

#include "dblogger.h"

#include <getopt.h>

#include "config.h"
#include "log.h"
#include "sqlite_storage.h"

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

// From LSB
#define EXIT_INVALIDARGUMENT 2 // Invalid or excess arguments
#define EXIT_NOTCONFIGURED 6   // The program is not configured

using namespace std;
using namespace std::chrono;

namespace
{
    //! Maximum timeout before forced application termination. Topic cleanup can
    //! take a lot of time
    const auto DRIVER_STOP_TIMEOUT_S = chrono::seconds(5);

    //! Maximun time to start application. Exceded timeout will case application
    //! termination.
    const auto DRIVER_INIT_TIMEOUT_S = chrono::seconds(30);

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

    void PrintUsage()
    {
        cout << "Usage:" << endl
             << " " << APP_NAME << " [options]" << endl
             << "Options:" << endl
             << "  -d       level     enable debuging output:" << endl
             << "                       1 - db 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;
    }

    void ParseCommadLine(int argc, char* argv[], WBMQTT::TMosquittoMqttConfig& mqttConfig, string& config)
    {
        int debugLevel = 0;
        int c;
        while ((c = getopt(argc, argv, "d:c:h:H:p:u:P:T:")) != -1) {
            switch (c) {
                case 'd':
                    debugLevel = stoi(optarg);
                    break;
                case 'c':
                    config = 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;

                case '?':
                default:
                    PrintUsage();
                    exit(EXIT_INVALIDARGUMENT);
            }
        }

        switch (debugLevel) {
            case 0:
                break;
            case -1:
                Info.SetEnabled(false);
                break;

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

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

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

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

            case 3:
                WBMQTT::Debug.SetEnabled(true);
                Debug.SetEnabled(true);
                break;

            default:
                cout << "Invalid -d parameter value " << debugLevel << endl;
                PrintUsage();
                exit(EXIT_INVALIDARGUMENT);
        }

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

    void PrintStartupInfo(const WBMQTT::TMosquittoMqttConfig& mqttConfig, const string& configFile)
    {
        cout << "MQTT broker " << mqttConfig.Host << ':' << mqttConfig.Port << endl;
        cout << "Config file " << configFile << endl;
    }

} // namespace

int main(int argc, char* argv[])
{
    WBMQTT::TMosquittoMqttConfig mqttConfig;
    mqttConfig.Id = APP_NAME;

    string configFileName("/etc/wb-mqtt-db.conf");

    ParseCommadLine(argc, argv, mqttConfig, configFileName);
    PrintStartupInfo(mqttConfig, configFileName);

    WBMQTT::TPromise<void> initialized;
    WBMQTT::SetThreadName("main");
    WBMQTT::SignalHandling::Handle({SIGINT, SIGTERM});
    WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [&] { WBMQTT::SignalHandling::Stop(); });

    /* if signal arrived before driver is initialized:
        wait some time to initialize and then exit gracefully
        else if timed out: exit with error
    */
    WBMQTT::SignalHandling::SetWaitFor(DRIVER_INIT_TIMEOUT_S, initialized.GetFuture(), [&] {
        Error.Log() << "Driver takes too long to initialize. Exiting.";
        cerr << "Error: DRIVER_INIT_TIMEOUT_S" << endl;
        exit(1);
    });

    /* if handling of signal takes too much time: exit with error */
    WBMQTT::SignalHandling::SetOnTimeout(DRIVER_STOP_TIMEOUT_S, [&] {
        Error.Log() << "Driver takes too long to stop. Exiting.";
        cerr << "Error: DRIVER_STOP_TIMEOUT_S" << endl;
        exit(2);
    });

    TMQTTDBLoggerConfig config;
    try {
        config = LoadConfig(configFileName, "/usr/share/wb-mqtt-confed/schemas/wb-mqtt-db.schema.json");
    } catch (const std::exception& e) {
        Error.Log() << e.what();
        return EXIT_NOTCONFIGURED;
    }

    WBMQTT::SignalHandling::Start();
    try {
        if (config.Debug)
            Debug.SetEnabled(true);

        auto mqttClient(WBMQTT::NewMosquittoMqttClient(mqttConfig));
        auto backend = WBMQTT::NewDriverBackend(mqttClient);
        auto driver = WBMQTT::NewDriver(WBMQTT::TDriverArgs{}.SetId(APP_NAME).SetBackend(backend));
        auto rpcServer = WBMQTT::NewMqttRpcServer(mqttClient, "db_logger");
        std::shared_ptr<TMQTTDBLogger> logger(new TMQTTDBLogger(driver,
                                                                config.Cache,
                                                                std::make_unique<TSqliteStorage>(config.DBFile),
                                                                rpcServer,
                                                                std::make_unique<TChannelWriter>(),
                                                                config.GetValuesRpcRequestTimeout));

        WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [=] { logger->Stop(); });
        initialized.Complete();
        Info.Log() << "DB logger started, go to main loop";
        logger->Start();
    } catch (const std::exception& e) {
        Error.Log() << e.what();
        WBMQTT::SignalHandling::Stop();
        return EXIT_FAILURE;
    }
    WBMQTT::SignalHandling::Wait();
    return EXIT_SUCCESS;
}

1	#include "dblogger.h"
2
3	#include <getopt.h>
4
5	#include "config.h"
6	#include "log.h"
7	#include "sqlite_storage.h"
8
9	#include <wblib/rpc.h>
10	#include <wblib/signal_handling.h>
11	#include <wblib/wbmqtt.h>
12
13	// From LSB
14	#define EXIT_INVALIDARGUMENT 2 // Invalid or excess arguments
15	#define EXIT_NOTCONFIGURED 6 // The program is not configured
16
17	using namespace std;
18	using namespace std::chrono;
19
20	namespace
21	{
22	//! Maximum timeout before forced application termination. Topic cleanup can
23	//! take a lot of time
24	const auto DRIVER_STOP_TIMEOUT_S = chrono::seconds(5);
25
26	//! Maximun time to start application. Exceded timeout will case application
27	//! termination.
28	const auto DRIVER_INIT_TIMEOUT_S = chrono::seconds(30);
29
30	const auto APP_NAME = "wb-mqtt-db";
31
32	void PrintUsage()	×
33	{
34	cout << "Usage:" << endl	×
35	<< " " << APP_NAME << " [options]" << endl	×
36	<< "Options:" << endl	×
37	<< " -d level enable debuging output:" << endl	×
38	<< " 1 - db only;" << endl	×
39	<< " 2 - mqtt only;" << endl	×
40	<< " 3 - both;" << endl	×
41	<< " negative values - silent mode (-1, -2, -3))" << endl	×
42	<< " -c config config file" << endl	×
43	<< " -p port MQTT broker port (default: 1883)" << endl	×
44	<< " -h, -H IP MQTT broker IP (default: localhost)" << endl	×
45	<< " -u user MQTT user (optional)" << endl	×
46	<< " -P password MQTT user password (optional)" << endl	×
47	<< " -T prefix MQTT topic prefix (optional)" << endl;	×
48	}	×
49
50	void ParseCommadLine(int argc, char* argv[], WBMQTT::TMosquittoMqttConfig& mqttConfig, string& config)	×
51	{
52	int debugLevel = 0;	×
53	int c;
54	while ((c = getopt(argc, argv, "d:c:h:H:p:u:P:T:")) != -1) {	×
55	switch (c) {	×
56	case 'd':	×
57	debugLevel = stoi(optarg);	×
58	break;	×
59	case 'c':	×
60	config = optarg;	×
61	break;	×
62	case 'p':	×
63	mqttConfig.Port = stoi(optarg);	×
64	break;	×
65	case 'h':	×
66	case 'H': // backward compatibility
67	mqttConfig.Host = optarg;	×
68	break;	×
69	case 'T':	×
70	mqttConfig.Prefix = optarg;	×
71	break;	×
72	case 'u':	×
73	mqttConfig.User = optarg;	×
74	break;	×
75	case 'P':	×
76	mqttConfig.Password = optarg;	×
77	break;	×
78
79	case '?':	×
80	default:
81	PrintUsage();	×
82	exit(EXIT_INVALIDARGUMENT);	×
83	}
84	}
85
86	switch (debugLevel) {	×
87	case 0:	×
88	break;	×
89	case -1:	×
90	Info.SetEnabled(false);	×
91	break;	×
92
93	case -2:	×
94	WBMQTT::Info.SetEnabled(false);	×
95	break;	×
96
97	case -3:	×
98	WBMQTT::Info.SetEnabled(false);	×
99	Info.SetEnabled(false);	×
100	break;	×
101
102	case 1:	×
103	Debug.SetEnabled(true);	×
104	break;	×
105
106	case 2:	×
107	WBMQTT::Debug.SetEnabled(true);	×
108	break;	×
109
110	case 3:	×
111	WBMQTT::Debug.SetEnabled(true);	×
112	Debug.SetEnabled(true);	×
113	break;	×
114
115	default:	×
116	cout << "Invalid -d parameter value " << debugLevel << endl;	×
117	PrintUsage();	×
118	exit(EXIT_INVALIDARGUMENT);	×
119	}
120
121	if (optind < argc) {	×
122	for (int index = optind; index < argc; ++index) {	×
123	cout << "Skipping unknown argument " << argv[index] << endl;	×
124	}
125	}
126	}	×
127
128	void PrintStartupInfo(const WBMQTT::TMosquittoMqttConfig& mqttConfig, const string& configFile)	×
129	{
130	cout << "MQTT broker " << mqttConfig.Host << ':' << mqttConfig.Port << endl;	×
131	cout << "Config file " << configFile << endl;	×
132	}	×
133
134	} // namespace
135
136	int main(int argc, char* argv[])	×
137	{
138	WBMQTT::TMosquittoMqttConfig mqttConfig;	×
139	mqttConfig.Id = APP_NAME;	×
140
141	string configFileName("/etc/wb-mqtt-db.conf");	×
142
143	ParseCommadLine(argc, argv, mqttConfig, configFileName);	×
144	PrintStartupInfo(mqttConfig, configFileName);	×
145
146	WBMQTT::TPromise<void> initialized;	×
147	WBMQTT::SetThreadName("main");	×
148	WBMQTT::SignalHandling::Handle({SIGINT, SIGTERM});	×
149	WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [&] { WBMQTT::SignalHandling::Stop(); });	×
150
151	/* if signal arrived before driver is initialized:
152	wait some time to initialize and then exit gracefully
153	else if timed out: exit with error
154	*/
155	WBMQTT::SignalHandling::SetWaitFor(DRIVER_INIT_TIMEOUT_S, initialized.GetFuture(), [&] {	×
156	Error.Log() << "Driver takes too long to initialize. Exiting.";	×
157	cerr << "Error: DRIVER_INIT_TIMEOUT_S" << endl;	×
158	exit(1);	×
159	});
160
161	/* if handling of signal takes too much time: exit with error */
162	WBMQTT::SignalHandling::SetOnTimeout(DRIVER_STOP_TIMEOUT_S, [&] {	×
163	Error.Log() << "Driver takes too long to stop. Exiting.";	×
164	cerr << "Error: DRIVER_STOP_TIMEOUT_S" << endl;	×
165	exit(2);	×
166	});
167
168	TMQTTDBLoggerConfig config;	×
169	try {
170	config = LoadConfig(configFileName, "/usr/share/wb-mqtt-confed/schemas/wb-mqtt-db.schema.json");	×
171	} catch (const std::exception& e) {	×
172	Error.Log() << e.what();	×
173	return EXIT_NOTCONFIGURED;	×
174	}	×
175
176	WBMQTT::SignalHandling::Start();	×
177	try {
178	if (config.Debug)	×
179	Debug.SetEnabled(true);	×
180
181	auto mqttClient(WBMQTT::NewMosquittoMqttClient(mqttConfig));	×
182	auto backend = WBMQTT::NewDriverBackend(mqttClient);	×
183	auto driver = WBMQTT::NewDriver(WBMQTT::TDriverArgs{}.SetId(APP_NAME).SetBackend(backend));	×
184	auto rpcServer = WBMQTT::NewMqttRpcServer(mqttClient, "db_logger");	×
185	std::shared_ptr<TMQTTDBLogger> logger(new TMQTTDBLogger(driver,
186	config.Cache,
187	std::make_unique<TSqliteStorage>(config.DBFile),	×
188	rpcServer,
189	std::make_unique<TChannelWriter>(),	×
190	config.GetValuesRpcRequestTimeout));	×
191
192	WBMQTT::SignalHandling::OnSignals({SIGINT, SIGTERM}, [=] { logger->Stop(); });	×
193	initialized.Complete();	×
194	Info.Log() << "DB logger started, go to main loop";	×
195	logger->Start();	×
196	} catch (const std::exception& e) {	×
197	Error.Log() << e.what();	×
198	WBMQTT::SignalHandling::Stop();	×
199	return EXIT_FAILURE;	×
200	}	×
201	WBMQTT::SignalHandling::Wait();	×
202	return EXIT_SUCCESS;	×
203	}	×

wirenboard / wb-mqtt-db / 96

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