wirenboard / wb-mqtt-mbgate / 5

Committed 20 Aug 2026 01:18PM UTC coverage: 69.312% (-0.04%) from 69.352%

Build # 5

Build Type

push

github

Committed by

ekateluv

Commit Message

Align service behaviour with the controller service guideline

- Exit with 7 on SIGTERM/SIGINT, 6 on malformed config file
- Log the all-channels-disabled message as Info
- systemd unit: RestartSec=1

Coverage Stats

706 of 927 branches covered (76.16%)

0 of 4 new or added lines in 2 files covered. (0.0%)

1188 of 1714 relevant lines covered (69.31%)

11.39 hits per line

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

0.0

/src/config_parser.cpp

#include "config_parser.h"

#include <fstream>
#include <iostream>
#include <memory>
#include <string>
#include <time.h>
#include <tuple>
#include <unistd.h>

#include <wblib/json_utils.h>
#include <wblib/mqtt.h>

#include "log.h"
#include "mbgate_exception.h"
#include "modbus_lmb_backend.h"
#include "mqtt_converters.h"
#include "observer.h"

using namespace std;
using namespace WBMQTT;
using namespace WBMQTT::JSON;

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

namespace
{
    string expandTopic(const string& t)
    {
        auto lst = StringSplit(t, '/');
        return string("/devices/") + lst[0] + "/controls/" + lst[1];
    }
};

IConfigParser::~IConfigParser()
{}

TJSONConfigParser::TJSONConfigParser(const string& config_file, const string& schema_file)
{
    try {
        Root = Parse(config_file);
        auto schema(Parse(schema_file));
        Validate(Root, schema);
    } catch (const std::runtime_error& e) {
        throw TConfigException(e.what());
    }
}

bool TJSONConfigParser::Debug()
{
    bool debug = false;
    Get(Root, "debug", debug);
    return debug;
}

tuple<PModbusServer, PMqttClient> TJSONConfigParser::Build()
{
    PModbusBackend modbusBackend = nullptr;

    auto modbus_data = Root["modbus"];

    if (modbus_data.isMember("path")) {
        TModbusRTUBackendArgs args{};

        args.Device = modbus_data["path"].asCString();

        args.BaudRate = modbus_data.get("baud_rate", args.BaudRate).asInt();
        args.DataBits = modbus_data.get("data_bits", args.DataBits).asInt();
        args.StopBits = modbus_data.get("stop_bits", args.StopBits).asInt();
        args.Parity = modbus_data.get("parity", std::string(1, args.Parity)).asString()[0];

        LOG(Debug) << "Modbus configuration: device " << args.Device << ", baud rate " << args.BaudRate << ", parity "
                   << args.Parity << ", data bits " << args.DataBits << ", stop bits " << args.StopBits;

        modbusBackend = make_shared<TModbusRTUBackend>(args);
    } else {
        string modbus_host = modbus_data["host"].asString();
        int modbus_port = modbus_data["port"].asInt();

        LOG(Debug) << "Modbus configuration: host " << modbus_host << ", port " << modbus_port;
        modbusBackend = make_shared<TModbusTCPBackend>(modbus_host.c_str(), modbus_port);
    }

    PModbusServer modbus = make_shared<TModbusServer>(modbusBackend);

    // create MQTT client
    string mqtt_host = Root["mqtt"]["host"].asString();
    int mqtt_port = Root["mqtt"]["port"].asInt();
    int mqtt_keepalive = 60;
    Get(Root["mqtt"], "keepalive", mqtt_keepalive);

    LOG(Debug) << "MQTT configuration: host " << mqtt_host << ", port " << mqtt_port << ", keepalive "
               << mqtt_keepalive;

    TMosquittoMqttConfig mqtt_config;
    mqtt_config.Host = mqtt_host;
    mqtt_config.Port = mqtt_port;
    mqtt_config.Keepalive = mqtt_keepalive;
    mqtt_config.Id = string("mqtt-mbgate-") + to_string(time(NULL));

    PMqttClient mqtt = NewMosquittoMqttClient(mqtt_config);

    bool any_enabled = false;

    // create observers and link'em with MQTT and Modbus
    any_enabled |= _BuildStore(COIL, Root["registers"]["coils"], modbus, mqtt);
    any_enabled |= _BuildStore(DISCRETE_INPUT, Root["registers"]["discretes"], modbus, mqtt);
    any_enabled |= _BuildStore(HOLDING_REGISTER, Root["registers"]["holdings"], modbus, mqtt);
    any_enabled |= _BuildStore(INPUT_REGISTER, Root["registers"]["inputs"], modbus, mqtt);

    if (!any_enabled) {
        throw TEmptyConfigException();
    }

    return make_tuple(modbus, mqtt);
}

bool TJSONConfigParser::_BuildStore(TStoreType type, const Json::Value& list, PModbusServer modbus, PMqttClient mqtt)
{
    LOG(Debug) << "Processing store " << type;

    bool enabled = false;

    for (const auto& reg_item: list) {
        if (!reg_item["enabled"].asBool()) {
            continue;
        }
        enabled = true;

        int address = reg_item["address"].asInt();
        int slave_id = reg_item["unitId"].asInt();
        string topic = reg_item["topic"].asString();
        int size;

        LOG(Debug) << "Element " << topic << " : " << address;

        PGatewayObserver obs;
        PMQTTConverter conv;

        if (type == COIL || type == DISCRETE_INPUT) {
            conv = make_shared<TMQTTDiscrConverter>();
            size = 1;
        } else {
            string format = reg_item["format"].asString();
            bool byteswap = reg_item["byteswap"].asBool();
            bool wordswap = reg_item["wordswap"].asBool();
            size = reg_item["size"].asInt();

            if (format == "varchar") {
                // old version config may contain varchar registers with negative size value (-1)
                // set this registers size to 1
                if (size < 0)
                    size = 1;
                conv = make_shared<TMQTTTextConverter>(size, byteswap, wordswap);
            } else if (format == "float") {
                double scale = reg_item["scale"].asFloat();
                conv = make_shared<TMQTTFloatConverter>(size, byteswap, wordswap, scale);
                size /= 2;
            } else {
                double scale = reg_item["scale"].asFloat();

                TMQTTIntConverter::IntegerType int_type;
                if (format == "signed") {
                    int_type = TMQTTIntConverter::SIGNED;
                } else if (format == "unsigned") {
                    int_type = TMQTTIntConverter::UNSIGNED;
                } else if (format == "bcd") {
                    int_type = TMQTTIntConverter::BCD;
                } else {
                    throw TConfigException("Unknown integer format: " + format);
                }

                conv = make_shared<TMQTTIntConverter>(int_type, scale, size, byteswap, wordswap);
                size /= 2;
            }
        }

        obs = make_shared<TGatewayObserver>(expandTopic(topic), conv, mqtt);

        LOG(Debug) << "Creating observer on " << address << ":" << size;

        try {
            modbus->Observe(obs, type, TModbusAddressRange(address, size), slave_id);
        } catch (const WrongSegmentException& e) {
            throw TConfigException(string("Address overlapping: ") + StoreTypeToString(type) + ": topic " + topic);
        }
    }
    return enabled;
}

1	#include "config_parser.h"
2
3	#include <fstream>
4	#include <iostream>
5	#include <memory>
6	#include <string>
7	#include <time.h>
8	#include <tuple>
9	#include <unistd.h>
10
11	#include <wblib/json_utils.h>
12	#include <wblib/mqtt.h>
13
14	#include "log.h"
15	#include "mbgate_exception.h"
16	#include "modbus_lmb_backend.h"
17	#include "mqtt_converters.h"
18	#include "observer.h"
19
20	using namespace std;
21	using namespace WBMQTT;
22	using namespace WBMQTT::JSON;
23
24	#define LOG(logger) ::logger.Log() << "[config] "
25
26	namespace
27	{
28	string expandTopic(const string& t)	×
29	{
30	auto lst = StringSplit(t, '/');	×
31	return string("/devices/") + lst[0] + "/controls/" + lst[1];	×
32	}	×
33	};
34
35	IConfigParser::~IConfigParser()	×
36	{}	×
37
NEW 38	TJSONConfigParser::TJSONConfigParser(const string& config_file, const string& schema_file)	×
39	{
40	try {
NEW 41	Root = Parse(config_file);	×
42	auto schema(Parse(schema_file));	×
43	Validate(Root, schema);	×
44	} catch (const std::runtime_error& e) {	×
45	throw TConfigException(e.what());	×
46	}	×
47	}	×
48
49	bool TJSONConfigParser::Debug()	×
50	{
51	bool debug = false;	×
52	Get(Root, "debug", debug);	×
53	return debug;	×
54	}
55
56	tuple<PModbusServer, PMqttClient> TJSONConfigParser::Build()	×
57	{
58	PModbusBackend modbusBackend = nullptr;	×
59
60	auto modbus_data = Root["modbus"];	×
61
62	if (modbus_data.isMember("path")) {	×
63	TModbusRTUBackendArgs args{};	×
64
65	args.Device = modbus_data["path"].asCString();	×
66
67	args.BaudRate = modbus_data.get("baud_rate", args.BaudRate).asInt();	×
68	args.DataBits = modbus_data.get("data_bits", args.DataBits).asInt();	×
69	args.StopBits = modbus_data.get("stop_bits", args.StopBits).asInt();	×
70	args.Parity = modbus_data.get("parity", std::string(1, args.Parity)).asString()[0];	×
71
72	LOG(Debug) << "Modbus configuration: device " << args.Device << ", baud rate " << args.BaudRate << ", parity "	×
73	<< args.Parity << ", data bits " << args.DataBits << ", stop bits " << args.StopBits;	×
74
75	modbusBackend = make_shared<TModbusRTUBackend>(args);	×
76	} else {	×
77	string modbus_host = modbus_data["host"].asString();	×
78	int modbus_port = modbus_data["port"].asInt();	×
79
80	LOG(Debug) << "Modbus configuration: host " << modbus_host << ", port " << modbus_port;	×
81	modbusBackend = make_shared<TModbusTCPBackend>(modbus_host.c_str(), modbus_port);	×
82	}	×
83
84	PModbusServer modbus = make_shared<TModbusServer>(modbusBackend);	×
85
86	// create MQTT client
87	string mqtt_host = Root["mqtt"]["host"].asString();	×
88	int mqtt_port = Root["mqtt"]["port"].asInt();	×
89	int mqtt_keepalive = 60;	×
90	Get(Root["mqtt"], "keepalive", mqtt_keepalive);	×
91
92	LOG(Debug) << "MQTT configuration: host " << mqtt_host << ", port " << mqtt_port << ", keepalive "	×
93	<< mqtt_keepalive;	×
94
95	TMosquittoMqttConfig mqtt_config;	×
96	mqtt_config.Host = mqtt_host;	×
97	mqtt_config.Port = mqtt_port;	×
98	mqtt_config.Keepalive = mqtt_keepalive;	×
99	mqtt_config.Id = string("mqtt-mbgate-") + to_string(time(NULL));	×
100
101	PMqttClient mqtt = NewMosquittoMqttClient(mqtt_config);	×
102
103	bool any_enabled = false;	×
104
105	// create observers and link'em with MQTT and Modbus
106	any_enabled \|= _BuildStore(COIL, Root["registers"]["coils"], modbus, mqtt);	×
107	any_enabled \|= _BuildStore(DISCRETE_INPUT, Root["registers"]["discretes"], modbus, mqtt);	×
108	any_enabled \|= _BuildStore(HOLDING_REGISTER, Root["registers"]["holdings"], modbus, mqtt);	×
109	any_enabled \|= _BuildStore(INPUT_REGISTER, Root["registers"]["inputs"], modbus, mqtt);	×
110
111	if (!any_enabled) {	×
112	throw TEmptyConfigException();	×
113	}
114
115	return make_tuple(modbus, mqtt);	×
116	}	×
117
118	bool TJSONConfigParser::_BuildStore(TStoreType type, const Json::Value& list, PModbusServer modbus, PMqttClient mqtt)	×
119	{
120	LOG(Debug) << "Processing store " << type;	×
121
122	bool enabled = false;	×
123
124	for (const auto& reg_item: list) {	×
125	if (!reg_item["enabled"].asBool()) {	×
126	continue;	×
127	}
128	enabled = true;	×
129
130	int address = reg_item["address"].asInt();	×
131	int slave_id = reg_item["unitId"].asInt();	×
132	string topic = reg_item["topic"].asString();	×
133	int size;
134
135	LOG(Debug) << "Element " << topic << " : " << address;	×
136
137	PGatewayObserver obs;	×
138	PMQTTConverter conv;	×
139
140	if (type == COIL \|\| type == DISCRETE_INPUT) {	×
141	conv = make_shared<TMQTTDiscrConverter>();	×
142	size = 1;	×
143	} else {
144	string format = reg_item["format"].asString();	×
145	bool byteswap = reg_item["byteswap"].asBool();	×
146	bool wordswap = reg_item["wordswap"].asBool();	×
147	size = reg_item["size"].asInt();	×
148
149	if (format == "varchar") {	×
150	// old version config may contain varchar registers with negative size value (-1)
151	// set this registers size to 1
152	if (size < 0)	×
153	size = 1;	×
154	conv = make_shared<TMQTTTextConverter>(size, byteswap, wordswap);	×
155	} else if (format == "float") {	×
156	double scale = reg_item["scale"].asFloat();	×
157	conv = make_shared<TMQTTFloatConverter>(size, byteswap, wordswap, scale);	×
158	size /= 2;	×
159	} else {
160	double scale = reg_item["scale"].asFloat();	×
161
162	TMQTTIntConverter::IntegerType int_type;
163	if (format == "signed") {	×
164	int_type = TMQTTIntConverter::SIGNED;	×
165	} else if (format == "unsigned") {	×
166	int_type = TMQTTIntConverter::UNSIGNED;	×
167	} else if (format == "bcd") {	×
168	int_type = TMQTTIntConverter::BCD;	×
169	} else {
170	throw TConfigException("Unknown integer format: " + format);	×
171	}
172
173	conv = make_shared<TMQTTIntConverter>(int_type, scale, size, byteswap, wordswap);	×
174	size /= 2;	×
175	}
176	}	×
177
178	obs = make_shared<TGatewayObserver>(expandTopic(topic), conv, mqtt);	×
179
180	LOG(Debug) << "Creating observer on " << address << ":" << size;	×
181
182	try {
183	modbus->Observe(obs, type, TModbusAddressRange(address, size), slave_id);	×
184	} catch (const WrongSegmentException& e) {	×
185	throw TConfigException(string("Address overlapping: ") + StoreTypeToString(type) + ": topic " + topic);	×
186	}	×
187	}	×
188	return enabled;	×
189	}

wirenboard / wb-mqtt-mbgate / 5

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