4 * (c) 2020--2022 Martin Mares <mj@ucw.cz>
8 #include <ucw/fastbuf.h>
11 #include <ucw/string.h>
21 #include <curl/curl.h>
22 #include <mosquitto.h>
24 #define LOG_DIR "/var/log/bsb"
28 static struct mosquitto *mosq;
30 static void mqtt_error(const char *operation, int err, bool teardown)
32 msg(L_ERROR, "Mosquitto: %s failed: error %d", operation, err);
35 static void mqtt_publish(const char *topic, const char *fmt, ...)
41 int l = vsnprintf(m, sizeof(m), fmt, args);
42 int err = mosquitto_publish(mosq, NULL, topic, l, m, 0, true);
43 if (err != MOSQ_ERR_SUCCESS)
44 mqtt_error("publish", err, false);
49 static void mqtt_log_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int level, const char *message)
51 // msg(L_INFO, "MQTT(%d): %s", level, message);
54 static void mqtt_conn_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, int status)
57 msg(L_DEBUG, "MQTT: Connection established, subscribing");
58 int err = mosquitto_subscribe(mosq, NULL, "bsb/#", 1);
59 if (err != MOSQ_ERR_SUCCESS) {
60 mqtt_error("subscribe", err, false);
63 mqtt_publish("status/bsb-logger", "ok");
65 msg(L_DEBUG, "MQTT: Cannot connect");
69 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m);
71 static void mqtt_connect(void)
77 mosq = mosquitto_new("bsb-logger", 1, NULL);
79 die("Mosquitto: initialization failed");
81 mosquitto_connect_callback_set(mosq, mqtt_conn_callback);
82 mosquitto_log_callback_set(mosq, mqtt_log_callback);
83 mosquitto_message_callback_set(mosq, mqtt_msg_callback);
85 if (mosquitto_will_set(mosq, "status/bsb-logger", 4, "dead", 0, true) != MOSQ_ERR_SUCCESS)
86 die("Mosquitto: unable to set will");
88 if (mosquitto_tls_set(mosq, "/etc/burrow-mqtt/ca.crt", NULL, "/etc/burrow-mqtt/client.crt", "/etc/burrow-mqtt/client.key", NULL) != MOSQ_ERR_SUCCESS)
89 die("Mosquitto: unable to set TLS parameters");
91 if (mosquitto_connect_async(mosq, "burrow-mqtt", 8883, 60) != MOSQ_ERR_SUCCESS)
92 die("Mosquitto: connect failed");
95 static struct log_stream *packet_log, *stat_log;
97 static void mqtt_msg_callback(struct mosquitto *mosq UNUSED, void *obj UNUSED, const struct mosquitto_message *m)
99 time_t now = time(NULL);
101 if (m->payloadlen >= sizeof(val) - 1) {
102 msg(L_ERROR, "Invalid value for topic %s", m->topic);
105 memcpy(val, m->payload, m->payloadlen);
106 val[m->payloadlen] = 0;
107 msg(L_DEBUG, "MQTT < %s %s", m->topic, val);
109 if (!strcmp(m->topic, "bsb/frame"))
110 msg(L_INFO | packet_log->regnum, "%s", val);
111 else if (str_has_prefix(m->topic, "bsb/stats/"))
112 msg(L_INFO | stat_log->regnum, "%s:%s", m->topic + 10, val);
117 static int use_daemon;
118 static int use_debug;
120 static struct opt_section options = {
122 OPT_HELP("A daemon for logging BSB frames received via MQTT"),
124 OPT_HELP("Options:"),
125 OPT_BOOL('d', "debug", use_debug, 0, "\tLog debugging messages"),
126 OPT_BOOL(0, "daemon", use_daemon, 0, "\tDaemonize"),
133 int main(int argc UNUSED, char **argv)
136 opt_parse(&options, argv+1);
139 struct log_stream *ls = log_new_syslog("daemon", LOG_PID);
140 log_set_default_stream(ls);
143 log_default_stream()->levels &= ~(1U << L_DEBUG);
145 packet_log = log_new_file(LOG_DIR "/frames-%Y%m%d", 0);
146 packet_log->msgfmt = LSFMT_TIME;
147 stat_log = log_new_file(LOG_DIR "/stats-%Y%m%d", 0);
148 stat_log->msgfmt = LSFMT_TIME;
152 int err = mosquitto_loop_forever(mosq, 10000, 1);
153 if (err != MOSQ_ERR_SUCCESS)
154 mqtt_error("loop", err, false);