X-Git-Url: https://git.octo.it/?p=collectd.git;a=blobdiff_plain;f=src%2Fmqtt.c;h=48c34edc036897e7853599bfc6b0a6e197be2419;hp=315974b6f49a8b6b7d85a59427de1f0cc9a6026f;hb=ec51ddee94fa2ba1e01fe0e336ccc9c190a198ff;hpb=f374b72032a227a75b6bc9ae574cd28abbc16f24

diff --git a/src/mqtt.c b/src/mqtt.c
index 315974b6..48c34edc 100644
--- a/src/mqtt.c
+++ b/src/mqtt.c
@@ -54,11 +54,11 @@
  * Data types
  */
 struct mqtt_client_conf {
-  _Bool publish;
+  bool publish;
   char *name;
 
   struct mosquitto *mosq;
-  _Bool connected;
+  bool connected;
 
   char *host;
   int port;
@@ -74,22 +74,22 @@ struct mqtt_client_conf {
 
   /* For publishing */
   char *topic_prefix;
-  _Bool store_rates;
-  _Bool retain;
+  bool store_rates;
+  bool retain;
 
   /* For subscribing */
   pthread_t thread;
-  _Bool loop;
+  bool loop;
   char *topic;
-  _Bool clean_session;
+  bool clean_session;
 
   c_complain_t complaint_cantpublish;
   pthread_mutex_t lock;
 };
 typedef struct mqtt_client_conf mqtt_client_conf_t;
 
-static mqtt_client_conf_t **subscribers = NULL;
-static size_t subscribers_num = 0;
+static mqtt_client_conf_t **subscribers;
+static size_t subscribers_num;
 
 /*
  * Functions
@@ -141,7 +141,7 @@ static void mqtt_free(mqtt_client_conf_t *conf) {
 
   if (conf->connected)
     (void)mosquitto_disconnect(conf->mosq);
-  conf->connected = 0;
+  conf->connected = false;
   (void)mosquitto_destroy(conf->mosq);
 
   sfree(conf->host);
@@ -160,17 +160,17 @@ static char *strip_prefix(char *topic) {
       num++;
 
   if (num < 2)
-    return (NULL);
+    return NULL;
 
   while (num > 2) {
     char *tmp = strchr(topic, '/');
     if (tmp == NULL)
-      return (NULL);
+      return NULL;
     topic = tmp + 1;
     num--;
   }
 
-  return (topic);
+  return topic;
 }
 
 static void on_message(
@@ -243,24 +243,22 @@ static int mqtt_reconnect(mqtt_client_conf_t *conf) {
   int status;
 
   if (conf->connected)
-    return (0);
+    return 0;
 
   status = mosquitto_reconnect(conf->mosq);
   if (status != MOSQ_ERR_SUCCESS) {
-    char errbuf[1024];
     ERROR("mqtt_connect_broker: mosquitto_connect failed: %s",
-          (status == MOSQ_ERR_ERRNO) ? sstrerror(errno, errbuf, sizeof(errbuf))
-                                     : mosquitto_strerror(status));
-    return (-1);
+          (status == MOSQ_ERR_ERRNO) ? STRERRNO : mosquitto_strerror(status));
+    return -1;
   }
 
-  conf->connected = 1;
+  conf->connected = true;
 
   c_release(LOG_INFO, &conf->complaint_cantpublish,
             "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
             conf->host, conf->port);
 
-  return (0);
+  return 0;
 } /* mqtt_reconnect */
 
 /* must hold conf->lock when calling. */
@@ -284,7 +282,7 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
 #endif
   if (conf->mosq == NULL) {
     ERROR("mqtt plugin: mosquitto_new failed");
-    return (-1);
+    return -1;
   }
 
 #if LIBMOSQUITTO_MAJOR != 0
@@ -297,7 +295,7 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
             mosquitto_strerror(status));
       mosquitto_destroy(conf->mosq);
       conf->mosq = NULL;
-      return (-1);
+      return -1;
     }
 
     status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER,
@@ -307,7 +305,7 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
             mosquitto_strerror(status));
       mosquitto_destroy(conf->mosq);
       conf->mosq = NULL;
-      return (-1);
+      return -1;
     }
 
     status = mosquitto_tls_insecure_set(conf->mosq, false);
@@ -316,7 +314,7 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
             mosquitto_strerror(status));
       mosquitto_destroy(conf->mosq);
       conf->mosq = NULL;
-      return (-1);
+      return -1;
     }
   }
 #endif
@@ -325,15 +323,12 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
     status =
         mosquitto_username_pw_set(conf->mosq, conf->username, conf->password);
     if (status != MOSQ_ERR_SUCCESS) {
-      char errbuf[1024];
       ERROR("mqtt plugin: mosquitto_username_pw_set failed: %s",
-            (status == MOSQ_ERR_ERRNO)
-                ? sstrerror(errno, errbuf, sizeof(errbuf))
-                : mosquitto_strerror(status));
+            (status == MOSQ_ERR_ERRNO) ? STRERRNO : mosquitto_strerror(status));
 
       mosquitto_destroy(conf->mosq);
       conf->mosq = NULL;
-      return (-1);
+      return -1;
     }
   }
 
@@ -346,14 +341,12 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
       mosquitto_connect(conf->mosq, conf->host, conf->port, MQTT_KEEPALIVE);
 #endif
   if (status != MOSQ_ERR_SUCCESS) {
-    char errbuf[1024];
     ERROR("mqtt plugin: mosquitto_connect failed: %s",
-          (status == MOSQ_ERR_ERRNO) ? sstrerror(errno, errbuf, sizeof(errbuf))
-                                     : mosquitto_strerror(status));
+          (status == MOSQ_ERR_ERRNO) ? STRERRNO : mosquitto_strerror(status));
 
     mosquitto_destroy(conf->mosq);
     conf->mosq = NULL;
-    return (-1);
+    return -1;
   }
 
   if (!conf->publish) {
@@ -369,12 +362,12 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
       mosquitto_disconnect(conf->mosq);
       mosquitto_destroy(conf->mosq);
       conf->mosq = NULL;
-      return (-1);
+      return -1;
     }
   }
 
-  conf->connected = 1;
-  return (0);
+  conf->connected = true;
+  return 0;
 } /* mqtt_connect */
 
 static void *subscribers_thread(void *arg) {
@@ -400,14 +393,14 @@ static void *subscribers_thread(void *arg) {
                             /* max_packets = */ 100);
 #endif
     if (status == MOSQ_ERR_CONN_LOST) {
-      conf->connected = 0;
+      conf->connected = false;
       continue;
     } else if (status != MOSQ_ERR_SUCCESS) {
       ERROR("mqtt plugin: mosquitto_loop failed: %s",
             mosquitto_strerror(status));
       mosquitto_destroy(conf->mosq);
       conf->mosq = NULL;
-      conf->connected = 0;
+      conf->connected = false;
       continue;
     }
 
@@ -427,7 +420,7 @@ static int publish(mqtt_client_conf_t *conf, char const *topic,
   if (status != 0) {
     pthread_mutex_unlock(&conf->lock);
     ERROR("mqtt plugin: unable to reconnect to broker");
-    return (status);
+    return status;
   }
 
   status = mosquitto_publish(conf->mosq, /* message_id */ NULL, topic,
@@ -438,24 +431,22 @@ static int publish(mqtt_client_conf_t *conf, char const *topic,
 #endif
                              conf->qos, conf->retain);
   if (status != MOSQ_ERR_SUCCESS) {
-    char errbuf[1024];
     c_complain(LOG_ERR, &conf->complaint_cantpublish,
                "mqtt plugin: mosquitto_publish failed: %s",
-               (status == MOSQ_ERR_ERRNO)
-                   ? sstrerror(errno, errbuf, sizeof(errbuf))
-                   : mosquitto_strerror(status));
+               (status == MOSQ_ERR_ERRNO) ? STRERRNO
+                                          : 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->connected = 0;
+    conf->connected = false;
     mosquitto_disconnect(conf->mosq);
 
     pthread_mutex_unlock(&conf->lock);
-    return (-1);
+    return -1;
   }
 
   pthread_mutex_unlock(&conf->lock);
-  return (0);
+  return 0;
 } /* int publish */
 
 static int format_topic(char *buf, size_t buf_len, data_set_t const *ds,
@@ -465,21 +456,21 @@ static int format_topic(char *buf, size_t buf_len, data_set_t const *ds,
   char *c;
 
   if ((conf->topic_prefix == NULL) || (conf->topic_prefix[0] == 0))
-    return (FORMAT_VL(buf, buf_len, vl));
+    return FORMAT_VL(buf, buf_len, vl);
 
   status = FORMAT_VL(name, sizeof(name), vl);
   if (status != 0)
-    return (status);
+    return status;
 
-  status = ssnprintf(buf, buf_len, "%s/%s", conf->topic_prefix, name);
+  status = snprintf(buf, buf_len, "%s/%s", conf->topic_prefix, name);
   if ((status < 0) || (((size_t)status) >= buf_len))
-    return (ENOMEM);
+    return ENOMEM;
 
-  while((c = strchr(buf, '#')) || (c = strchr(buf, '+'))) {
-       *c = '_';
+  while ((c = strchr(buf, '#')) || (c = strchr(buf, '+'))) {
+    *c = '_';
   }
 
-  return (0);
+  return 0;
 } /* int format_topic */
 
 static int mqtt_write(const data_set_t *ds, const value_list_t *vl,
@@ -490,28 +481,28 @@ static int mqtt_write(const data_set_t *ds, const value_list_t *vl,
   int status = 0;
 
   if ((user_data == NULL) || (user_data->data == NULL))
-    return (EINVAL);
+    return EINVAL;
   conf = user_data->data;
 
   status = format_topic(topic, sizeof(topic), ds, vl, conf);
   if (status != 0) {
     ERROR("mqtt plugin: format_topic failed with status %d.", status);
-    return (status);
+    return status;
   }
 
   status = format_values(payload, sizeof(payload), ds, vl, conf->store_rates);
   if (status != 0) {
     ERROR("mqtt plugin: format_values failed with status %d.", status);
-    return (status);
+    return status;
   }
 
   status = publish(conf, topic, payload, strlen(payload) + 1);
   if (status != 0) {
     ERROR("mqtt plugin: publish failed: %s", mosquitto_strerror(status));
-    return (status);
+    return status;
   }
 
-  return (status);
+  return status;
 } /* mqtt_write */
 
 /*
@@ -525,10 +516,10 @@ static int mqtt_write(const data_set_t *ds, const value_list_t *vl,
  *   StoreRates true
  *   Retain false
  *   QoS 0
- *   CACert "ca.pem"			Enables TLS if set
- *   CertificateFile "client-cert.pem"		optional
- *   CertificateKeyFile "client-key.pem"		optional
- *   TLSProtocol "tlsv1.2"		optional
+ *   CACert "ca.pem"                      Enables TLS if set
+ *   CertificateFile "client-cert.pem"	  optional
+ *   CertificateKeyFile "client-key.pem"  optional
+ *   TLSProtocol "tlsv1.2"                optional
  * </Publish>
  */
 static int mqtt_config_publisher(oconfig_item_t *ci) {
@@ -539,15 +530,15 @@ static int mqtt_config_publisher(oconfig_item_t *ci) {
   conf = calloc(1, sizeof(*conf));
   if (conf == NULL) {
     ERROR("mqtt plugin: calloc failed.");
-    return (-1);
+    return -1;
   }
-  conf->publish = 1;
+  conf->publish = true;
 
   conf->name = NULL;
   status = cf_util_get_string(ci, &conf->name);
   if (status != 0) {
     mqtt_free(conf);
-    return (status);
+    return status;
   }
 
   conf->host = strdup(MQTT_DEFAULT_HOST);
@@ -555,12 +546,12 @@ static int mqtt_config_publisher(oconfig_item_t *ci) {
   conf->client_id = NULL;
   conf->qos = 0;
   conf->topic_prefix = strdup(MQTT_DEFAULT_TOPIC_PREFIX);
-  conf->store_rates = 1;
+  conf->store_rates = true;
 
   status = pthread_mutex_init(&conf->lock, NULL);
   if (status != 0) {
     mqtt_free(conf);
-    return (status);
+    return status;
   }
 
   C_COMPLAIN_INIT(&conf->complaint_cantpublish);
@@ -608,11 +599,12 @@ static int mqtt_config_publisher(oconfig_item_t *ci) {
       ERROR("mqtt plugin: Unknown config option: %s", child->key);
   }
 
-  ssnprintf(cb_name, sizeof(cb_name), "mqtt/%s", conf->name);
-  plugin_register_write(cb_name, mqtt_write, &(user_data_t){
-                                                 .data = conf,
-                                             });
-  return (0);
+  snprintf(cb_name, sizeof(cb_name), "mqtt/%s", conf->name);
+  plugin_register_write(cb_name, mqtt_write,
+                        &(user_data_t){
+                            .data = conf,
+                        });
+  return 0;
 } /* mqtt_config_publisher */
 
 /*
@@ -623,6 +615,10 @@ static int mqtt_config_publisher(oconfig_item_t *ci) {
  *   User "guest"
  *   Password "secret"
  *   Topic "collectd/#"
+ *   CACert "ca.pem"                      Enables TLS if set
+ *   CertificateFile "client-cert.pem"	  optional
+ *   CertificateKeyFile "client-key.pem"  optional
+ *   TLSProtocol "tlsv1.2"                optional
  * </Subscribe>
  */
 static int mqtt_config_subscriber(oconfig_item_t *ci) {
@@ -633,15 +629,15 @@ static int mqtt_config_subscriber(oconfig_item_t *ci) {
   conf = calloc(1, sizeof(*conf));
   if (conf == NULL) {
     ERROR("mqtt plugin: calloc failed.");
-    return (-1);
+    return -1;
   }
-  conf->publish = 0;
+  conf->publish = false;
 
   conf->name = NULL;
   status = cf_util_get_string(ci, &conf->name);
   if (status != 0) {
     mqtt_free(conf);
-    return (status);
+    return status;
   }
 
   conf->host = strdup(MQTT_DEFAULT_HOST);
@@ -649,12 +645,12 @@ static int mqtt_config_subscriber(oconfig_item_t *ci) {
   conf->client_id = NULL;
   conf->qos = 2;
   conf->topic = strdup(MQTT_DEFAULT_TOPIC);
-  conf->clean_session = 1;
+  conf->clean_session = true;
 
   status = pthread_mutex_init(&conf->lock, NULL);
   if (status != 0) {
     mqtt_free(conf);
-    return (status);
+    return status;
   }
 
   C_COMPLAIN_INIT(&conf->complaint_cantpublish);
@@ -686,6 +682,16 @@ static int mqtt_config_subscriber(oconfig_item_t *ci) {
       cf_util_get_string(child, &conf->topic);
     else if (strcasecmp("CleanSession", child->key) == 0)
       cf_util_get_boolean(child, &conf->clean_session);
+    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)
+      cf_util_get_string(child, &conf->tlsprotocol);
+    else if (strcasecmp("CipherSuite", child->key) == 0)
+      cf_util_get_string(child, &conf->ciphersuite);
     else
       ERROR("mqtt plugin: Unknown config option: %s", child->key);
   }
@@ -694,13 +700,13 @@ static int mqtt_config_subscriber(oconfig_item_t *ci) {
   if (tmp == NULL) {
     ERROR("mqtt plugin: realloc failed.");
     mqtt_free(conf);
-    return (-1);
+    return -1;
   }
   subscribers = tmp;
   subscribers[subscribers_num] = conf;
   subscribers_num++;
 
-  return (0);
+  return 0;
 } /* mqtt_config_subscriber */
 
 /*
@@ -725,7 +731,7 @@ static int mqtt_config(oconfig_item_t *ci) {
       ERROR("mqtt plugin: Unknown config option: %s", child->key);
   }
 
-  return (0);
+  return 0;
 } /* int mqtt_config */
 
 static int mqtt_init(void) {
@@ -743,19 +749,15 @@ static int mqtt_init(void) {
                                   /* args  = */ subscribers[i],
                                   /* name  = */ "mqtt");
     if (status != 0) {
-      char errbuf[1024];
-      ERROR("mqtt plugin: pthread_create failed: %s",
-            sstrerror(errno, errbuf, sizeof(errbuf)));
+      ERROR("mqtt plugin: pthread_create failed: %s", STRERRNO);
       continue;
     }
   }
 
-  return (0);
+  return 0;
 } /* mqtt_init */
 
 void module_register(void) {
   plugin_register_complex_config("mqtt", mqtt_config);
   plugin_register_init("mqtt", mqtt_init);
 } /* void module_register */
-
-/* vim: set sw=4 sts=4 et fdm=marker : */