X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fmqtt.c;h=dbef915ae4d681ba4f634e3042df52defdc7c609;hb=edd9af8a874ebc9f2a7f02846807229a648917db;hp=403b0d31e2a38ccdc2c26e5cf29516236249ee3b;hpb=69b0d4948ca934d22e6e88f1eca02678f0d45355;p=collectd.git

diff --git a/src/mqtt.c b/src/mqtt.c
index 403b0d31..dbef915a 100644
--- a/src/mqtt.c
+++ b/src/mqtt.c
@@ -36,8 +36,6 @@
 #include "utils_cache.h"
 #include "utils_complain.h"
 
-#include <pthread.h>
-
 #include <mosquitto.h>
 
 #define MQTT_MAX_TOPIC_SIZE         1024
@@ -185,9 +183,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 +215,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,10 +540,10 @@ 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
  * </Publish>
  */
 static int mqtt_config_publisher (oconfig_item_t *ci)
@@ -549,7 +557,7 @@ static int mqtt_config_publisher (oconfig_item_t *ci)
     conf = calloc (1, sizeof (*conf));
     if (conf == NULL)
     {
-        ERROR ("mqtt plugin: malloc failed.");
+        ERROR ("mqtt plugin: calloc failed.");
         return (-1);
     }
     conf->publish = 1;
@@ -612,13 +620,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);
@@ -642,7 +650,7 @@ static int mqtt_config_publisher (oconfig_item_t *ci)
  *   User "guest"
  *   Password "secret"
  *   Topic "collectd/#"
- * </Publish>
+ * </Subscribe>
  */
 static int mqtt_config_subscriber (oconfig_item_t *ci)
 {
@@ -654,7 +662,7 @@ static int mqtt_config_subscriber (oconfig_item_t *ci)
     conf = calloc (1, sizeof (*conf));
     if (conf == NULL)
     {
-        ERROR ("mqtt plugin: malloc failed.");
+        ERROR ("mqtt plugin: calloc failed.");
         return (-1);
     }
     conf->publish = 0;
@@ -690,11 +698,11 @@ static int mqtt_config_subscriber (oconfig_item_t *ci)
             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 +712,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 +727,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.");