X-Git-Url: https://git.verplant.org/?a=blobdiff_plain;f=src%2Fmqtt.c;h=dbef915ae4d681ba4f634e3042df52defdc7c609;hb=f91506bb8f91ca3725fd3081deba4e1829186aaa;hp=cbffa67e98d411f9c1fe805c1b75f0752cee8f66;hpb=aaff848cf8b7752c7549590637f1b7702e32221e;p=collectd.git

diff --git a/src/mqtt.c b/src/mqtt.c
index cbffa67e..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)
@@ -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;
@@ -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.");