From: Florian Forster Date: Thu, 20 Nov 2014 16:38:15 +0000 (+0100) Subject: configure.ac, src/Makefile.am: Add mqtt to the build system. X-Git-Tag: collectd-5.6.0~637^2~20 X-Git-Url: https://git.verplant.org/?a=commitdiff_plain;h=ed93534ffd9b64f72fc4d9dcff866191cddf3de6;p=collectd.git configure.ac, src/Makefile.am: Add mqtt to the build system. --- diff --git a/README b/README index 6ac7e2b9..ad5b6678 100644 --- a/README +++ b/README @@ -191,6 +191,9 @@ Features Reads values from Modbus/TCP enabled devices. Supports reading values from multiple "slaves" so gateway devices can be used. + - mqtt + Publishes and subscribes to MQTT topics. + - multimeter Information provided by serial multimeters, such as the `Metex M-4650CR'. diff --git a/README.md b/README.md deleted file mode 100644 index ef2b4deb..00000000 --- a/README.md +++ /dev/null @@ -1,38 +0,0 @@ -collectd-mqtt -============= - -This is my unfinished attempt to produce a MQTT write plugin for [collectd][0] -using the [Mosquitto][1] library. Due to a lack of time I never managed to get -the project to move forward. - -I'm not a professional developer – even less C developer, so expect this code to -be bad. Probably really bad. You've been warned. - -License -------- - -The MIT License (MIT) - -Copyright (c) 2014 Marc Falzon - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in -all copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -THE SOFTWARE. - - -[0]: https://collectd.org/ -[1]: http://mosquitto.org/ diff --git a/configure.ac b/configure.ac index 44d10cf6..6ee6e9a5 100644 --- a/configure.ac +++ b/configure.ac @@ -2653,6 +2653,53 @@ fi AM_CONDITIONAL(BUILD_WITH_LIBMONGOC, test "x$with_libmongoc" = "xyes") # }}} +# --with-libmosquitto {{{ +with_libmosquitto_cppflags="" +with_libmosquitto_libs="-lmosquitto" +AC_ARG_WITH(libmosquitto, [AS_HELP_STRING([--with-libmosquitto@<:@=PREFIX@:>@], [Path to libmosquitto.])], +[ + if test "x$withval" != "xno" && test "x$withval" != "xyes" + then + with_libmosquitto_cppflags="-I$withval/include" + with_libmosquitto_libs="-L$withval/lib -lmosquitto" + with_libmosquitto="yes" + else + with_libmosquitto="$withval" + fi +], +[ + with_libmosquitto="yes" +]) +if test "x$with_libmosquitto" = "xyes" +then + SAVE_CPPFLAGS="$CPPFLAGS" + CPPFLAGS="$with_libmosquitto_cppflags" + + AC_CHECK_HEADERS(mosquitto.h, [with_libmosquitto="yes"], [with_libmosquitto="no (mosquitto.h not found)"]) + + CPPFLAGS="$SAVE_CPPFLAGS" +fi +if test "x$with_libmosquitto" = "xyes" +then + SAVE_LDFLAGS="$LDFLAGS" + SAVE_CPPFLAGS="$CPPFLAGS" + LDFLAGS="$with_libmosquitto_libs" + CPPFLAGS="$with_libmosquitto_cppflags" + + AC_CHECK_LIB(mosquitto, mosquitto_connect, [with_libmosquitto="yes"], [with_libmosquitto="no (libmosquitto not found)"]) + + LDFLAGS="$SAVE_LDFLAGS" + CPPFLAGS="$SAVE_CPPFLAGS" +fi +if test "x$with_libmosquitto" = "xyes" +then + BUILD_WITH_LIBMOSQUITTO_CPPFLAGS="$with_libmosquitto_cppflags" + BUILD_WITH_LIBMOSQUITTO_LIBS="$with_libmosquitto_libs" + AC_SUBST(BUILD_WITH_LIBMOSQUITTO_CPPFLAGS) + AC_SUBST(BUILD_WITH_LIBMOSQUITTO_LIBS) +fi +# }}} + # --with-libmysql {{{ with_mysql_config="mysql_config" with_mysql_cflags="" @@ -5632,6 +5679,7 @@ AC_PLUGIN([memcached], [yes], [memcached statistics]) AC_PLUGIN([memory], [$plugin_memory], [Memory usage]) AC_PLUGIN([mic], [$with_mic], [Intel Many Integrated Core stats]) AC_PLUGIN([modbus], [$with_libmodbus], [Modbus plugin]) +AC_PLUGIN([mqtt], [$with_libmosquitto], [MQTT output plugin]) AC_PLUGIN([multimeter], [$plugin_multimeter], [Read multimeter values]) AC_PLUGIN([mysql], [$with_libmysql], [MySQL statistics]) AC_PLUGIN([netapp], [$with_libnetapp], [NetApp plugin]) @@ -5912,6 +5960,7 @@ Configuration: libmnl . . . . . . . $with_libmnl libmodbus . . . . . . $with_libmodbus libmongoc . . . . . . $with_libmongoc + libmosquitto . . . . $with_libmosquitto libmysql . . . . . . $with_libmysql libnetapp . . . . . . $with_libnetapp libnetsnmp . . . . . $with_libnetsnmp @@ -6011,6 +6060,7 @@ Configuration: memory . . . . . . . $enable_memory mic . . . . . . . . . $enable_mic modbus . . . . . . . $enable_modbus + mqtt . . . . . . . . $enable_mqtt multimeter . . . . . $enable_multimeter mysql . . . . . . . . $enable_mysql netapp . . . . . . . $enable_netapp diff --git a/mqtt.c b/mqtt.c deleted file mode 100644 index 3bed0a5e..00000000 --- a/mqtt.c +++ /dev/null @@ -1,370 +0,0 @@ -/** - * collectd - src/mqtt.c - * Copyright (C) 2014 Marc Falzon - * - * Permission is hereby granted, free of charge, to any person obtaining a - * copy of this software and associated documentation files (the "Software"), - * to deal in the Software without restriction, including without limitation - * the rights to use, copy, modify, merge, publish, distribute, sublicense, - * and/or sell copies of the Software, and to permit persons to whom the - * Software is furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING - * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER - * DEALINGS IN THE SOFTWARE. - **/ - -// Reference: http://mosquitto.org/api/files/mosquitto-h.html - - -#include "collectd.h" -#include "common.h" -#include "plugin.h" -#include "utils_cache.h" -#include "utils_complain.h" - -#include - -#include - -#define MQTT_MAX_TOPIC_SIZE 1024 -#define MQTT_MAX_MESSAGE_SIZE MQTT_MAX_TOPIC_SIZE + 1024 -#define MQTT_DEFAULT_HOST "localhost" -#define MQTT_DEFAULT_PORT 1883 -#define MQTT_DEFAULT_CLIENT_ID "collectd" -#define MQTT_DEFAULT_TOPIC_PREFIX "collectd" - -/* - * Data types - */ -struct mqtt_client_conf -{ - struct mosquitto *mosq; - bool connected; - char *host; - int port; - char *client_id; - char *topic_prefix; - c_complain_t complaint_cantpublish; - pthread_mutex_t lock; -}; - -/* - * Functions - */ -static int mqtt_reconnect_broker (struct mqtt_client_conf *conf) -{ - int status; - - if (conf->connected) - return (0); - - pthread_mutex_lock (&conf->lock); - - status = mosquitto_reconnect (conf->mosq); - - if (status != MOSQ_ERR_SUCCESS) { - ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s", - (status == MOSQ_ERR_ERRNO ? - strerror(errno) : mosquitto_strerror (status))); - pthread_mutex_unlock (&conf->lock); - return (-1); - } - - conf->connected = true; - - c_release (LOG_INFO, - &conf->complaint_cantpublish, - "mqtt plugin: successfully reconnected to broker \"%s:%d\"", - conf->host, conf->port); - - pthread_mutex_unlock (&conf->lock); - - return (0); -} /* mqtt_reconnect_broker */ - -static int mqtt_publish_message (struct mqtt_client_conf *conf, char *topic, - char *payload, size_t payload_len) -{ - char errbuf[1024]; - int status; - - status = mosquitto_publish(conf->mosq, - /* message id */ NULL, - topic, - (int) payload_len, - payload, - /* qos */ 0, - /* retain */ false); - - if (status != MOSQ_ERR_SUCCESS) - { - c_complain (LOG_ERR, - &conf->complaint_cantpublish, - "plugin mqtt: mosquitto_publish failed: %s", - status == MOSQ_ERR_ERRNO ? - sstrerror(errno, errbuf, sizeof (errbuf)) : - mosquitto_strerror(status)); - /* - Mark our connection "down" regardless of the error as a safety measure; - we will try to reconnect the next time we have to publish a message - */ - conf->connected = false; - - return (-1); - } - - return (0); -} /* mqtt_publish_message */ - -static int mqtt_format_metric_value (char *buf, size_t buf_len, - const data_set_t *data_set, const value_list_t *vl, int ds_num) -{ - gauge_t *rates = NULL; - gauge_t *value = NULL; - size_t metric_value_len; - int status = 0; - - memset (buf, 0, buf_len); - - if (data_set->ds[ds_num].type == DS_TYPE_GAUGE) - value = &vl->values[ds_num].gauge; - else { - rates = uc_get_rate (data_set, vl); - value = &rates[ds_num]; - } - - metric_value_len = ssnprintf (buf, buf_len, "%f", *value); - - if (metric_value_len >= buf_len) - return (-ENOMEM); - - if (rates) - sfree (rates); - - return (status); -} /* mqtt_format_metric_value */ - -static int mqtt_format_message_topic (char *buf, size_t buf_len, - char const *prefix, const value_list_t *vl, const char *ds_name) -{ - size_t topic_buf_len; - - memset (buf, 0, buf_len); - - /* - MQTT message topic format: - [/]////// - */ - topic_buf_len = (size_t) ssnprintf (buf, buf_len, - "%s/%s/%s/%s/%s/%s/%s", - prefix, - vl->host, - vl->plugin, - vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "(null)", - vl->type, - vl->type_instance[0] != '\0' ? vl->type_instance : "(null)", - ds_name); - - if (topic_buf_len >= buf_len) - { - ERROR ("mqtt_format_message_topic: topic buffer too small: " - "Need %zu bytes.", topic_buf_len + 1); - return (-ENOMEM); - } - - return (0); -} /* mqtt_format_message_topic */ - -static int mqtt_format_payload (char *buf, size_t buf_len, - const data_set_t *data_set, const value_list_t *vl, int ds_num) -{ - char metric_path[10 * DATA_MAX_NAME_LEN]; - char metric_value[512]; - size_t payload_buf_len; - int status = 0; - - memset (buf, 0, buf_len); - - ssnprintf (metric_path, sizeof (metric_path), - "%s.%s%s%s.%s%s%s%s%s", - vl->host, - vl->plugin, - vl->plugin_instance[0] != '\0' ? "." : "", - vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "", - vl->type, - vl->type_instance[0] != '\0' ? "." : "", - vl->type_instance[0] != '\0' ? vl->type_instance : "", - strcmp(data_set->ds[ds_num].name, "value") != 0 ? "." : "", - strcmp(data_set->ds[ds_num].name, "value") != 0 ? - data_set->ds[ds_num].name : ""); - - status = mqtt_format_metric_value (metric_value, - sizeof (metric_value), - data_set, - vl, - ds_num); - - if (status != 0) - { - ERROR ("mqtt_format_payload: error with mqtt_format_metric_value"); - return (status); - } - - payload_buf_len = (size_t) ssnprintf (buf, buf_len, - "%s %s %u", - metric_path, - metric_value, - (unsigned int) CDTIME_T_TO_TIME_T (vl->time)); - - if (payload_buf_len >= buf_len) - { - ERROR ("mqtt_format_payload: payload buffer too small: " - "Need %zu bytes.", payload_buf_len + 1); - return (-ENOMEM); - } - - return (status); -} /* mqtt_format_payload */ - -static int mqtt_write (const data_set_t *data_set, const value_list_t *vl, - user_data_t *user_data) -{ - struct mqtt_client_conf *conf; - char msg_topic[MQTT_MAX_TOPIC_SIZE]; - char msg_payload[MQTT_MAX_MESSAGE_SIZE]; - int status = 0; - int i; - - if (user_data == NULL) - return (EINVAL); - - conf = user_data->data; - - if (!conf->connected) - { - status = mqtt_reconnect_broker (conf); - - if (status != 0) { - ERROR ("plugin mqtt: unable to reconnect to broker"); - return (status); - } - } - - for (i = 0; i < data_set->ds_num; i++) - { - status = mqtt_format_message_topic (msg_topic, sizeof (msg_topic), - conf->topic_prefix, vl, data_set->ds[i].name); - if (status != 0) - { - ERROR ("plugin mqtt: error with mqtt_format_message_topic"); - return (status); - } - - status = mqtt_format_payload (msg_payload, - sizeof (msg_payload), - data_set, - vl, - i); - - if (status != 0) - { - ERROR ("mqtt_write: error with mqtt_format_payload"); - return (status); - } - - status = mqtt_publish_message (conf, - msg_topic, - msg_payload, - sizeof (msg_payload)); - if (status != 0) - { - ERROR ("plugin mqtt: unable to publish message"); - return (status); - } - - DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_write[%02X]\x1B[0m " - "published message: topic=%s payload=%s", - (unsigned)pthread_self(), - msg_topic, - msg_payload); - } - - return (status); -} /* mqtt_write */ - -static int mqtt_config (oconfig_item_t *ci) -{ - struct mqtt_client_conf *conf; - user_data_t user_data; - char errbuf[1024]; - int status; - - DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_config[%02X]\x1B[0m ", - (unsigned)pthread_self()); - - conf = malloc (sizeof (*conf)); - if (conf == NULL) - { - ERROR ("write_mqtt plugin: malloc failed."); - return (-1); - } - - memset (conf, 0, sizeof (*conf)); - - conf->connected = false; - conf->host = MQTT_DEFAULT_HOST; - conf->port = MQTT_DEFAULT_PORT; - conf->client_id = MQTT_DEFAULT_CLIENT_ID; - conf->topic_prefix = MQTT_DEFAULT_TOPIC_PREFIX; - C_COMPLAIN_INIT (&conf->complaint_cantpublish); - - memset (&user_data, 0, sizeof (user_data)); - user_data.data = conf; - - if ((conf->mosq = mosquitto_new (conf->client_id, true, NULL)) == NULL) { - ERROR ("mqtt_config: mosquitto_new failed"); - return (-1); - } - - status = mosquitto_connect (conf->mosq, conf->host, conf->port, 10); - - if (status != MOSQ_ERR_SUCCESS) { - ERROR ("mqtt_config: mosquitto_connect failed: %s", - (status == MOSQ_ERR_ERRNO ? - sstrerror(errno, errbuf, sizeof (errbuf)) : - mosquitto_strerror (status))); - return (-1); - } - - DEBUG ("mqtt plugin: successfully connected to broker \"%s:%d\"", - conf->host, conf->port); - - conf->connected = true; - - plugin_register_write ("mqtt", mqtt_write, &user_data); - - return (0); -} /* mqtt_config */ - -static int mqtt_init (void) -{ - mosquitto_lib_init(); - - return (0); -} /* mqtt_init */ - -void module_register (void) -{ - plugin_register_complex_config ("mqtt", mqtt_config); - plugin_register_init ("mqtt", mqtt_init); -} /* void module_register */ - -/* vim: set sw=4 sts=4 et fdm=marker : */ diff --git a/mqtt.conf b/mqtt.conf deleted file mode 100644 index eae25bd2..00000000 --- a/mqtt.conf +++ /dev/null @@ -1,6 +0,0 @@ - - - Host "localhost" - TopicPrefix "collectd" - - diff --git a/src/Makefile.am b/src/Makefile.am index 2c7c1821..0ca9b718 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -637,6 +637,14 @@ modbus_la_CFLAGS = $(AM_CFLAGS) $(BUILD_WITH_LIBMODBUS_CFLAGS) modbus_la_LIBADD = $(BUILD_WITH_LIBMODBUS_LIBS) endif +if BUILD_PLUGIN_MQTT +pkglib_LTLIBRARIES += mqtt.la +mqtt_la_SOURCES = mqtt.c +mqtt_la_LDFLAGS = $(PLUGIN_LDFLAGS) +mqtt_la_CFLAGS = $(AM_CFLAGS) $(BUILD_WITH_LIBMOSQUITTO_CFLAGS) +mqtt_la_LIBADD = $(BUILD_WITH_LIBMOSQUITTO_LIBS) +endif + if BUILD_PLUGIN_MULTIMETER pkglib_LTLIBRARIES += multimeter.la multimeter_la_SOURCES = multimeter.c diff --git a/src/mqtt.c b/src/mqtt.c new file mode 100644 index 00000000..3bed0a5e --- /dev/null +++ b/src/mqtt.c @@ -0,0 +1,370 @@ +/** + * collectd - src/mqtt.c + * Copyright (C) 2014 Marc Falzon + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + **/ + +// Reference: http://mosquitto.org/api/files/mosquitto-h.html + + +#include "collectd.h" +#include "common.h" +#include "plugin.h" +#include "utils_cache.h" +#include "utils_complain.h" + +#include + +#include + +#define MQTT_MAX_TOPIC_SIZE 1024 +#define MQTT_MAX_MESSAGE_SIZE MQTT_MAX_TOPIC_SIZE + 1024 +#define MQTT_DEFAULT_HOST "localhost" +#define MQTT_DEFAULT_PORT 1883 +#define MQTT_DEFAULT_CLIENT_ID "collectd" +#define MQTT_DEFAULT_TOPIC_PREFIX "collectd" + +/* + * Data types + */ +struct mqtt_client_conf +{ + struct mosquitto *mosq; + bool connected; + char *host; + int port; + char *client_id; + char *topic_prefix; + c_complain_t complaint_cantpublish; + pthread_mutex_t lock; +}; + +/* + * Functions + */ +static int mqtt_reconnect_broker (struct mqtt_client_conf *conf) +{ + int status; + + if (conf->connected) + return (0); + + pthread_mutex_lock (&conf->lock); + + status = mosquitto_reconnect (conf->mosq); + + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s", + (status == MOSQ_ERR_ERRNO ? + strerror(errno) : mosquitto_strerror (status))); + pthread_mutex_unlock (&conf->lock); + return (-1); + } + + conf->connected = true; + + c_release (LOG_INFO, + &conf->complaint_cantpublish, + "mqtt plugin: successfully reconnected to broker \"%s:%d\"", + conf->host, conf->port); + + pthread_mutex_unlock (&conf->lock); + + return (0); +} /* mqtt_reconnect_broker */ + +static int mqtt_publish_message (struct mqtt_client_conf *conf, char *topic, + char *payload, size_t payload_len) +{ + char errbuf[1024]; + int status; + + status = mosquitto_publish(conf->mosq, + /* message id */ NULL, + topic, + (int) payload_len, + payload, + /* qos */ 0, + /* retain */ false); + + if (status != MOSQ_ERR_SUCCESS) + { + c_complain (LOG_ERR, + &conf->complaint_cantpublish, + "plugin mqtt: mosquitto_publish failed: %s", + status == MOSQ_ERR_ERRNO ? + sstrerror(errno, errbuf, sizeof (errbuf)) : + mosquitto_strerror(status)); + /* + Mark our connection "down" regardless of the error as a safety measure; + we will try to reconnect the next time we have to publish a message + */ + conf->connected = false; + + return (-1); + } + + return (0); +} /* mqtt_publish_message */ + +static int mqtt_format_metric_value (char *buf, size_t buf_len, + const data_set_t *data_set, const value_list_t *vl, int ds_num) +{ + gauge_t *rates = NULL; + gauge_t *value = NULL; + size_t metric_value_len; + int status = 0; + + memset (buf, 0, buf_len); + + if (data_set->ds[ds_num].type == DS_TYPE_GAUGE) + value = &vl->values[ds_num].gauge; + else { + rates = uc_get_rate (data_set, vl); + value = &rates[ds_num]; + } + + metric_value_len = ssnprintf (buf, buf_len, "%f", *value); + + if (metric_value_len >= buf_len) + return (-ENOMEM); + + if (rates) + sfree (rates); + + return (status); +} /* mqtt_format_metric_value */ + +static int mqtt_format_message_topic (char *buf, size_t buf_len, + char const *prefix, const value_list_t *vl, const char *ds_name) +{ + size_t topic_buf_len; + + memset (buf, 0, buf_len); + + /* + MQTT message topic format: + [/]////// + */ + topic_buf_len = (size_t) ssnprintf (buf, buf_len, + "%s/%s/%s/%s/%s/%s/%s", + prefix, + vl->host, + vl->plugin, + vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "(null)", + vl->type, + vl->type_instance[0] != '\0' ? vl->type_instance : "(null)", + ds_name); + + if (topic_buf_len >= buf_len) + { + ERROR ("mqtt_format_message_topic: topic buffer too small: " + "Need %zu bytes.", topic_buf_len + 1); + return (-ENOMEM); + } + + return (0); +} /* mqtt_format_message_topic */ + +static int mqtt_format_payload (char *buf, size_t buf_len, + const data_set_t *data_set, const value_list_t *vl, int ds_num) +{ + char metric_path[10 * DATA_MAX_NAME_LEN]; + char metric_value[512]; + size_t payload_buf_len; + int status = 0; + + memset (buf, 0, buf_len); + + ssnprintf (metric_path, sizeof (metric_path), + "%s.%s%s%s.%s%s%s%s%s", + vl->host, + vl->plugin, + vl->plugin_instance[0] != '\0' ? "." : "", + vl->plugin_instance[0] != '\0' ? vl->plugin_instance : "", + vl->type, + vl->type_instance[0] != '\0' ? "." : "", + vl->type_instance[0] != '\0' ? vl->type_instance : "", + strcmp(data_set->ds[ds_num].name, "value") != 0 ? "." : "", + strcmp(data_set->ds[ds_num].name, "value") != 0 ? + data_set->ds[ds_num].name : ""); + + status = mqtt_format_metric_value (metric_value, + sizeof (metric_value), + data_set, + vl, + ds_num); + + if (status != 0) + { + ERROR ("mqtt_format_payload: error with mqtt_format_metric_value"); + return (status); + } + + payload_buf_len = (size_t) ssnprintf (buf, buf_len, + "%s %s %u", + metric_path, + metric_value, + (unsigned int) CDTIME_T_TO_TIME_T (vl->time)); + + if (payload_buf_len >= buf_len) + { + ERROR ("mqtt_format_payload: payload buffer too small: " + "Need %zu bytes.", payload_buf_len + 1); + return (-ENOMEM); + } + + return (status); +} /* mqtt_format_payload */ + +static int mqtt_write (const data_set_t *data_set, const value_list_t *vl, + user_data_t *user_data) +{ + struct mqtt_client_conf *conf; + char msg_topic[MQTT_MAX_TOPIC_SIZE]; + char msg_payload[MQTT_MAX_MESSAGE_SIZE]; + int status = 0; + int i; + + if (user_data == NULL) + return (EINVAL); + + conf = user_data->data; + + if (!conf->connected) + { + status = mqtt_reconnect_broker (conf); + + if (status != 0) { + ERROR ("plugin mqtt: unable to reconnect to broker"); + return (status); + } + } + + for (i = 0; i < data_set->ds_num; i++) + { + status = mqtt_format_message_topic (msg_topic, sizeof (msg_topic), + conf->topic_prefix, vl, data_set->ds[i].name); + if (status != 0) + { + ERROR ("plugin mqtt: error with mqtt_format_message_topic"); + return (status); + } + + status = mqtt_format_payload (msg_payload, + sizeof (msg_payload), + data_set, + vl, + i); + + if (status != 0) + { + ERROR ("mqtt_write: error with mqtt_format_payload"); + return (status); + } + + status = mqtt_publish_message (conf, + msg_topic, + msg_payload, + sizeof (msg_payload)); + if (status != 0) + { + ERROR ("plugin mqtt: unable to publish message"); + return (status); + } + + DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_write[%02X]\x1B[0m " + "published message: topic=%s payload=%s", + (unsigned)pthread_self(), + msg_topic, + msg_payload); + } + + return (status); +} /* mqtt_write */ + +static int mqtt_config (oconfig_item_t *ci) +{ + struct mqtt_client_conf *conf; + user_data_t user_data; + char errbuf[1024]; + int status; + + DEBUG ("\x1B[36m[debug]\x1B[0m\x1B[37m mqtt_config[%02X]\x1B[0m ", + (unsigned)pthread_self()); + + conf = malloc (sizeof (*conf)); + if (conf == NULL) + { + ERROR ("write_mqtt plugin: malloc failed."); + return (-1); + } + + memset (conf, 0, sizeof (*conf)); + + conf->connected = false; + conf->host = MQTT_DEFAULT_HOST; + conf->port = MQTT_DEFAULT_PORT; + conf->client_id = MQTT_DEFAULT_CLIENT_ID; + conf->topic_prefix = MQTT_DEFAULT_TOPIC_PREFIX; + C_COMPLAIN_INIT (&conf->complaint_cantpublish); + + memset (&user_data, 0, sizeof (user_data)); + user_data.data = conf; + + if ((conf->mosq = mosquitto_new (conf->client_id, true, NULL)) == NULL) { + ERROR ("mqtt_config: mosquitto_new failed"); + return (-1); + } + + status = mosquitto_connect (conf->mosq, conf->host, conf->port, 10); + + if (status != MOSQ_ERR_SUCCESS) { + ERROR ("mqtt_config: mosquitto_connect failed: %s", + (status == MOSQ_ERR_ERRNO ? + sstrerror(errno, errbuf, sizeof (errbuf)) : + mosquitto_strerror (status))); + return (-1); + } + + DEBUG ("mqtt plugin: successfully connected to broker \"%s:%d\"", + conf->host, conf->port); + + conf->connected = true; + + plugin_register_write ("mqtt", mqtt_write, &user_data); + + return (0); +} /* mqtt_config */ + +static int mqtt_init (void) +{ + mosquitto_lib_init(); + + return (0); +} /* mqtt_init */ + +void module_register (void) +{ + plugin_register_complex_config ("mqtt", mqtt_config); + plugin_register_init ("mqtt", mqtt_init); +} /* void module_register */ + +/* vim: set sw=4 sts=4 et fdm=marker : */