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);
}
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)
{
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 */
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++)
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++)