X-Git-Url: https://git.verplant.org/?a=blobdiff_plain;f=src%2Fmqtt.c;h=32304f0cec888633796588274893ba069f77fd2f;hb=1d888f3f3a131f8eb2d9e0884665b6951ecee550;hp=403b0d31e2a38ccdc2c26e5cf29516236249ee3b;hpb=69b0d4948ca934d22e6e88f1eca02678f0d45355;p=collectd.git diff --git a/src/mqtt.c b/src/mqtt.c index 403b0d31..32304f0c 100644 --- a/src/mqtt.c +++ b/src/mqtt.c @@ -31,13 +31,11 @@ #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 @@ -147,11 +145,9 @@ static void mqtt_free (mqtt_client_conf_t *conf) static char *strip_prefix (char *topic) { - size_t num; - size_t i; + size_t num = 0; - num = 0; - for (i = 0; topic[i] != 0; i++) + for (size_t i = 0; topic[i] != 0; i++) if (topic[i] == '/') num++; @@ -185,9 +181,10 @@ static void on_message ( char *payload; int status; - if ((msg->payloadlen <= 0) - || (((uint8_t *) msg->payload)[msg->payloadlen - 1] != 0)) + if (msg->payloadlen <= 0) { + DEBUG ("mqtt plugin: message has empty payload"); return; + } topic = strdup (msg->topic); name = strip_prefix (topic); @@ -216,7 +213,16 @@ static void on_message ( } vl.values_len = ds->ds_num; - payload = strdup ((void *) msg->payload); + payload = malloc (msg->payloadlen+1); + if (payload == NULL) + { + ERROR ("mqtt plugin: malloc for payload buffer failed."); + sfree (vl.values); + return; + } + memmove (payload, msg->payload, msg->payloadlen); + payload[msg->payloadlen] = 0; + DEBUG ("mqtt plugin: payload = \"%s\"", payload); status = parse_values (payload, &vl, ds); if (status != 0) @@ -532,24 +538,22 @@ static int mqtt_write (const data_set_t *ds, const value_list_t *vl, * StoreRates true * Retain false * QoS 0 - * CACertificateFile "ca.pem" Enables TLS if set + * CACert "ca.pem" Enables TLS if set * CertificateFile "client-cert.pem" optional * CertificateKeyFile "client-key.pem" optional - * TLSprotocol "tlsv1.2" optional + * TLSProtocol "tlsv1.2" optional * */ static int mqtt_config_publisher (oconfig_item_t *ci) { mqtt_client_conf_t *conf; char cb_name[1024]; - user_data_t user_data; int status; - int i; conf = calloc (1, sizeof (*conf)); if (conf == NULL) { - ERROR ("mqtt plugin: malloc failed."); + ERROR ("mqtt plugin: calloc failed."); return (-1); } conf->publish = 1; @@ -578,7 +582,7 @@ static int mqtt_config_publisher (oconfig_item_t *ci) C_COMPLAIN_INIT (&conf->complaint_cantpublish); - for (i = 0; i < ci->children_num; i++) + for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; if (strcasecmp ("Host", child->key) == 0) @@ -612,13 +616,13 @@ static int mqtt_config_publisher (oconfig_item_t *ci) cf_util_get_boolean (child, &conf->store_rates); else if (strcasecmp ("Retain", child->key) == 0) cf_util_get_boolean (child, &conf->retain); - else if (strcasecmp ("CACertificateFile", child->key) == 0) + else if (strcasecmp ("CACert", child->key) == 0) cf_util_get_string (child, &conf->cacertificatefile); else if (strcasecmp ("CertificateFile", child->key) == 0) cf_util_get_string (child, &conf->certificatefile); else if (strcasecmp ("CertificateKeyFile", child->key) == 0) cf_util_get_string (child, &conf->certificatekeyfile); - else if (strcasecmp ("TLSprotocol", child->key) == 0) + else if (strcasecmp ("TLSProtocol", child->key) == 0) cf_util_get_string (child, &conf->tlsprotocol); else if (strcasecmp ("CipherSuite", child->key) == 0) cf_util_get_string (child, &conf->ciphersuite); @@ -627,10 +631,9 @@ static int mqtt_config_publisher (oconfig_item_t *ci) } ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name); - memset (&user_data, 0, sizeof (user_data)); - user_data.data = conf; - - plugin_register_write (cb_name, mqtt_write, &user_data); + plugin_register_write (cb_name, mqtt_write, &(user_data_t) { + .data = conf, + }); return (0); } /* mqtt_config_publisher */ @@ -642,19 +645,18 @@ static int mqtt_config_publisher (oconfig_item_t *ci) * User "guest" * Password "secret" * Topic "collectd/#" - * + * */ static int mqtt_config_subscriber (oconfig_item_t *ci) { mqtt_client_conf_t **tmp; mqtt_client_conf_t *conf; int status; - int i; conf = calloc (1, sizeof (*conf)); if (conf == NULL) { - ERROR ("mqtt plugin: malloc failed."); + ERROR ("mqtt plugin: calloc failed."); return (-1); } conf->publish = 0; @@ -683,18 +685,18 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) C_COMPLAIN_INIT (&conf->complaint_cantpublish); - for (i = 0; i < ci->children_num; i++) + for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; if (strcasecmp ("Host", child->key) == 0) cf_util_get_string (child, &conf->host); else if (strcasecmp ("Port", child->key) == 0) { - int tmp = cf_util_get_port_number (child); - if (tmp < 0) + status = cf_util_get_port_number (child); + if (status < 0) ERROR ("mqtt plugin: Invalid port number."); else - conf->port = tmp; + conf->port = status; } else if (strcasecmp ("ClientId", child->key) == 0) cf_util_get_string (child, &conf->client_id); @@ -704,12 +706,12 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) cf_util_get_string (child, &conf->password); else if (strcasecmp ("QoS", child->key) == 0) { - int tmp = -1; - status = cf_util_get_int (child, &tmp); - if ((status != 0) || (tmp < 0) || (tmp > 2)) + int qos = -1; + status = cf_util_get_int (child, &qos); + if ((status != 0) || (qos < 0) || (qos > 2)) ERROR ("mqtt plugin: Not a valid QoS setting."); else - conf->qos = tmp; + conf->qos = qos; } else if (strcasecmp ("Topic", child->key) == 0) cf_util_get_string (child, &conf->topic); @@ -719,7 +721,7 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) ERROR ("mqtt plugin: Unknown config option: %s", child->key); } - tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num); + tmp = realloc (subscribers, sizeof (*subscribers) * (subscribers_num + 1) ); if (tmp == NULL) { ERROR ("mqtt plugin: realloc failed."); @@ -745,9 +747,7 @@ static int mqtt_config_subscriber (oconfig_item_t *ci) */ static int mqtt_config (oconfig_item_t *ci) { - int i; - - for (i = 0; i < ci->children_num; i++) + for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; @@ -764,11 +764,9 @@ static int mqtt_config (oconfig_item_t *ci) static int mqtt_init (void) { - size_t i; - mosquitto_lib_init (); - for (i = 0; i < subscribers_num; i++) + for (size_t i = 0; i < subscribers_num; i++) { int status; @@ -778,7 +776,8 @@ static int mqtt_init (void) status = plugin_thread_create (&subscribers[i]->thread, /* attrs = */ NULL, /* func = */ subscribers_thread, - /* args = */ subscribers[i]); + /* args = */ subscribers[i], + /* name = */ "mqtt"); if (status != 0) { char errbuf[1024];