Merge remote-tracking branch 'github/pr/1355'
[collectd.git] / src / mqtt.c
index 210d38c..2eb1152 100644 (file)
@@ -176,9 +176,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);
@@ -207,7 +208,15 @@ 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.");
+        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)
@@ -407,11 +416,11 @@ static int publish (mqtt_client_conf_t *conf, char const *topic,
     {
         char errbuf[1024];
         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));
+            &conf->complaint_cantpublish,
+            "mqtt plugin: 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 */
@@ -527,6 +536,13 @@ static int mqtt_config_publisher (oconfig_item_t *ci)
     conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
     conf->store_rates = 1;
 
+    status = pthread_mutex_init (&conf->lock, NULL);
+    if (status != 0)
+    {
+      mqtt_free (conf);
+      return (status);
+    }
+
     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 
     for (i = 0; i < ci->children_num; i++)
@@ -615,6 +631,13 @@ static int mqtt_config_subscriber (oconfig_item_t *ci)
     conf->topic = strdup (MQTT_DEFAULT_TOPIC);
     conf->clean_session = 1;
 
+    status = pthread_mutex_init (&conf->lock, NULL);
+    if (status != 0)
+    {
+      mqtt_free (conf);
+      return (status);
+    }
+
     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 
     for (i = 0; i < ci->children_num; i++)