X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=src%2Fmqtt.c;h=b5e40aec4909563b96fffd4426ac28c3452e8017;hb=35a4e57c8e27e3ca1850c1dee291c19f1197e6bb;hp=c5254375867c38b90a5947fc61affcd54e2b0e02;hpb=b260f7243a91ea42baf0038bc703bb862f4d2ad5;p=collectd.git

diff --git a/src/mqtt.c b/src/mqtt.c
index c5254375..b5e40aec 100644
--- a/src/mqtt.c
+++ b/src/mqtt.c
@@ -1,6 +1,7 @@
 /**
  * collectd - src/mqtt.c
- * Copyright (C) 2014       Marc Falzon <marc at baha dot mu>
+ * Copyright (C) 2014       Marc Falzon
+ * Copyright (C) 2014,2015  Florian octo Forster
  *
  * Permission is hereby granted, free of charge, to any person obtaining a
  * copy of this software and associated documentation files (the "Software"),
@@ -19,6 +20,10 @@
  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  * DEALINGS IN THE SOFTWARE.
+ *
+ * Authors:
+ *   Marc Falzon <marc at baha dot mu>
+ *   Florian octo Forster <octo at collectd.org>
  **/
 
 // Reference: http://mosquitto.org/api/files/mosquitto-h.html
@@ -38,25 +43,53 @@
 #define MQTT_MAX_MESSAGE_SIZE       MQTT_MAX_TOPIC_SIZE + 1024
 #define MQTT_DEFAULT_HOST           "localhost"
 #define MQTT_DEFAULT_PORT           1883
-#define MQTT_DEFAULT_CLIENT_ID      "collectd"
 #define MQTT_DEFAULT_TOPIC_PREFIX   "collectd"
+#define MQTT_DEFAULT_TOPIC          "collectd/#"
+#ifndef MQTT_KEEPALIVE
+# define MQTT_KEEPALIVE 60
+#endif
+
 
 /*
  * Data types
  */
 struct mqtt_client_conf
 {
-    struct mosquitto    *mosq;
+    _Bool               publish;
+    char               *name;
+
+    struct mosquitto   *mosq;
     _Bool               connected;
-    char                *host;
+
+    char               *host;
     int                 port;
-    char                *client_id;
-    char                *topic_prefix;
+    char               *client_id;
+    char               *username;
+    char               *password;
+    int                 qos;
+
+    /* For publishing */
+    char               *topic_prefix;
+    _Bool               store_rates;
+    _Bool               retain;
+
+    /* For subscribing */
+    pthread_t           thread;
+    _Bool               loop;
+    char               *topic;
+    _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;
+
+/*
+ * Functions
+ */
 static char const *mosquitto_strerror (int code)
 {
     switch (code)
@@ -81,11 +114,107 @@ static char const *mosquitto_strerror (int code)
     return "UNKNOWN ERROR CODE";
 }
 
-/*
- * Functions
- */
+static void mqtt_free (mqtt_client_conf_t *conf)
+{
+    if (conf == NULL)
+        return;
+
+    if (conf->connected)
+        (void) mosquitto_disconnect (conf->mosq);
+    conf->connected = 0;
+    (void) mosquitto_destroy (conf->mosq);
+
+    sfree (conf->host);
+    sfree (conf->username);
+    sfree (conf->password);
+    sfree (conf->client_id);
+    sfree (conf->topic_prefix);
+    sfree (conf);
+}
+
+static char *strip_prefix (char *topic)
+{
+    size_t num;
+    size_t i;
+
+    num = 0;
+    for (i = 0; topic[i] != 0; i++)
+        if (topic[i] == '/')
+            num++;
+
+    if (num < 2)
+        return (NULL);
+
+    while (num > 2)
+    {
+        char *tmp = strchr (topic, '/');
+        if (tmp == NULL)
+            return (NULL);
+        topic = tmp + 1;
+        num--;
+    }
+
+    return (topic);
+}
+
+static void on_message (__attribute__((unused)) void *arg,
+        const struct mosquitto_message *msg)
+{
+    value_list_t vl = VALUE_LIST_INIT;
+    data_set_t const *ds;
+    char *topic;
+    char *name;
+    char *payload;
+    int status;
+
+    if ((msg->payloadlen <= 0) || (msg->payload[msg->payloadlen - 1] != 0))
+        return;
+
+    topic = strdup (msg->topic);
+    name = strip_prefix (topic);
+
+    status = parse_identifier_vl (name, &vl);
+    if (status != 0)
+    {
+        ERROR ("mqtt plugin: Unable to parse topic \"%s\".", topic);
+        sfree (topic);
+        return;
+    }
+    sfree (topic);
+
+    ds = plugin_get_ds (vl.type);
+    if (ds == NULL)
+    {
+        ERROR ("mqtt plugin: Unknown type: \"%s\".", vl.type);
+        return;
+    }
+
+    vl.values = calloc (ds->ds_num, sizeof (*vl.values));
+    if (vl.values == NULL)
+    {
+        ERROR ("mqtt plugin: calloc failed.");
+        return;
+    }
+    vl.values_len = ds->ds_num;
+
+    payload = strdup ((void *) msg->payload);
+    DEBUG ("mqtt plugin: payload = \"%s\"", payload);
+    status = parse_values (payload, &vl, ds);
+    if (status != 0)
+    {
+        ERROR ("mqtt plugin: Unable to parse payload \"%s\".", payload);
+        sfree (payload);
+        sfree (vl.values);
+        return;
+    }
+    sfree (payload);
+
+    plugin_dispatch_values (&vl);
+    sfree (vl.values);
+} /* void on_message */
+
 /* must hold conf->lock when calling. */
-static int mqtt_reconnect_broker (mqtt_client_conf_t *conf)
+static int mqtt_reconnect (mqtt_client_conf_t *conf)
 {
     int status;
 
@@ -111,17 +240,131 @@ static int mqtt_reconnect_broker (mqtt_client_conf_t *conf)
             conf->host, conf->port);
 
     return (0);
-} /* mqtt_reconnect_broker */
+} /* mqtt_reconnect */
+
+/* must hold conf->lock when calling. */
+static int mqtt_connect (mqtt_client_conf_t *conf)
+{
+    char const *client_id;
+    int status;
+
+    if (conf->mosq != NULL)
+        return mqtt_reconnect (conf);
+
+    if (conf->client_id)
+        client_id = conf->client_id;
+    else
+        client_id = hostname_g;
+
+    conf->mosq = mosquitto_new (client_id, /* user data = */ conf);
+    if (conf->mosq == NULL)
+    {
+        ERROR ("mqtt plugin: mosquitto_new failed");
+        return (-1);
+    }
+
+    if (conf->username && conf->password)
+    {
+        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));
+
+            mosquitto_destroy (conf->mosq);
+            conf->mosq = NULL;
+            return (-1);
+        }
+    }
+
+    status = mosquitto_connect (conf->mosq, conf->host, conf->port,
+            /* keepalive = */ MQTT_KEEPALIVE, /* clean session = */ conf->clean_session);
+    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));
+
+        mosquitto_destroy (conf->mosq);
+        conf->mosq = NULL;
+        return (-1);
+    }
+
+    if (!conf->publish)
+    {
+        mosquitto_message_callback_set (conf->mosq, on_message);
+
+        status = mosquitto_subscribe (conf->mosq, /* mid = */ NULL,
+                conf->topic, conf->qos);
+        if (status != MOSQ_ERR_SUCCESS)
+        {
+            ERROR ("mqtt plugin: Subscribing to \"%s\" failed: %s",
+                    conf->topic, mosquitto_strerror (status));
+
+            mosquitto_disconnect (conf->mosq);
+            mosquitto_destroy (conf->mosq);
+            conf->mosq = NULL;
+            return (-1);
+        }
+    }
+
+    conf->connected = 1;
+    return (0);
+} /* mqtt_connect */
+
+static void *subscribers_thread (void *arg)
+{
+    mqtt_client_conf_t *conf = arg;
+    int status;
+
+    conf->loop = 1;
+
+    while (conf->loop)
+    {
+        status = mqtt_connect (conf);
+        if (status != 0)
+        {
+            sleep (1);
+            continue;
+        }
+
+        /* The documentation says "0" would map to the default (1000ms), but
+         * that does not work on some versions. */
+        status = mosquitto_loop (conf->mosq, /* timeout = */ 1000 /* ms */);
+        if (status == MOSQ_ERR_CONN_LOST)
+        {
+            conf->connected = 0;
+            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;
+            continue;
+        }
+
+        DEBUG ("mqtt plugin: mosquitto_loop succeeded.");
+    } /* while (conf->loop) */
+
+    pthread_exit (0);
+} /* void *subscribers_thread */
 
 static int publish (mqtt_client_conf_t *conf, char const *topic,
     void const *payload, size_t payload_len)
 {
-    int const qos = 0; /* TODO: Config option */
     int status;
 
     pthread_mutex_lock (&conf->lock);
 
-    status = mqtt_reconnect_broker (conf);
+    status = mqtt_connect (conf);
     if (status != 0) {
         pthread_mutex_unlock (&conf->lock);
         ERROR ("mqtt plugin: unable to reconnect to broker");
@@ -132,8 +375,8 @@ static int publish (mqtt_client_conf_t *conf, char const *topic,
             /* message id */ NULL,
             topic,
             (uint32_t) payload_len, payload,
-            /* qos */ qos,
-            /* retain */ false);
+            /* qos */ conf->qos,
+            /* retain */ conf->retain);
     if (status != MOSQ_ERR_SUCCESS)
     {
         char errbuf[1024];
@@ -184,27 +427,27 @@ static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
     char topic[MQTT_MAX_TOPIC_SIZE];
     char payload[MQTT_MAX_MESSAGE_SIZE];
     int status = 0;
-    _Bool const store_rates = 0; /* TODO: Config option */
 
     if ((user_data == NULL) || (user_data->data == NULL))
         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);
     }
 
     status = format_values (payload, sizeof (payload),
-            ds, vl, store_rates);
+            ds, vl, conf->store_rates);
     if (status != 0)
     {
         ERROR ("mqtt plugin: format_values failed with status %d.", status);
         return (status);
     }
 
-    status = publish (conf, topic, payload, sizeof (payload));
+    status = publish (conf, topic, payload, strlen (payload) + 1);
     if (status != 0)
     {
         ERROR ("mqtt plugin: publish failed: %s", mosquitto_strerror (status));
@@ -215,16 +458,22 @@ static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
 } /* mqtt_write */
 
 /*
- * <Plugin mqtt>
+ * <Publish "name">
  *   Host "example.com"
  *   Port 1883
- *   Prefix "collectd"
  *   ClientId "collectd"
- * </Plugin>
+ *   User "guest"
+ *   Password "secret"
+ *   Prefix "collectd"
+ *   StoreRates true
+ *   Retain false
+ *   QoS 0
+ * </Publish>
  */
-static int mqtt_config (oconfig_item_t *ci)
+static int mqtt_config_publisher (oconfig_item_t *ci)
 {
     mqtt_client_conf_t *conf;
+    char cb_name[1024];
     user_data_t user_data;
     int status;
     int i;
@@ -235,12 +484,23 @@ static int mqtt_config (oconfig_item_t *ci)
         ERROR ("mqtt plugin: malloc failed.");
         return (-1);
     }
+    conf->publish = 1;
+
+    conf->name = NULL;
+    status = cf_util_get_string (ci, &conf->name);
+    if (status != 0)
+    {
+        mqtt_free (conf);
+        return (status);
+    }
 
-    conf->connected = 0;
     conf->host = strdup (MQTT_DEFAULT_HOST);
     conf->port = MQTT_DEFAULT_PORT;
-    conf->client_id = strdup (MQTT_DEFAULT_CLIENT_ID);
+    conf->client_id = NULL;
+    conf->qos = 0;
     conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
+    conf->store_rates = 1;
+
     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 
     for (i = 0; i < ci->children_num; i++)
@@ -252,57 +512,189 @@ static int mqtt_config (oconfig_item_t *ci)
         {
             int tmp = cf_util_get_port_number (child);
             if (tmp < 0)
-            {
                 ERROR ("mqtt plugin: Invalid port number.");
-                continue;
-            }
-            conf->port = tmp;
+            else
+                conf->port = tmp;
         }
-        else if (strcasecmp ("Prefix", child->key) == 0)
-            cf_util_get_string (child, &conf->topic_prefix);
         else if (strcasecmp ("ClientId", child->key) == 0)
             cf_util_get_string (child, &conf->client_id);
+        else if (strcasecmp ("User", child->key) == 0)
+            cf_util_get_string (child, &conf->username);
+        else if (strcasecmp ("Password", child->key) == 0)
+            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))
+                ERROR ("mqtt plugin: Not a valid QoS setting.");
+            else
+                conf->qos = tmp;
+        }
+        else if (strcasecmp ("Prefix", child->key) == 0)
+            cf_util_get_string (child, &conf->topic_prefix);
+        else if (strcasecmp ("StoreRates", child->key) == 0)
+            cf_util_get_boolean (child, &conf->store_rates);
+        else if (strcasecmp ("Retain", child->key) == 0)
+            cf_util_get_boolean (child, &conf->retain);
         else
             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
     }
 
+    ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name);
     memset (&user_data, 0, sizeof (user_data));
     user_data.data = conf;
 
-    conf->mosq = mosquitto_new (conf->client_id, /* user data = */ conf);
-    if (conf->mosq == NULL)
+    plugin_register_write (cb_name, mqtt_write, &user_data);
+    return (0);
+} /* mqtt_config_publisher */
+
+/*
+ * <Subscribe "name">
+ *   Host "example.com"
+ *   Port 1883
+ *   ClientId "collectd"
+ *   User "guest"
+ *   Password "secret"
+ *   Topic "collectd/#"
+ * </Publish>
+ */
+static int mqtt_config_subscriber (oconfig_item_t *ci)
+{
+    mqtt_client_conf_t **tmp;
+    mqtt_client_conf_t *conf;
+    int status;
+    int i;
+
+    conf = calloc (1, sizeof (*conf));
+    if (conf == NULL)
     {
-        ERROR ("mqtt plugin: mosquitto_new failed");
-        free (conf);
+        ERROR ("mqtt plugin: malloc failed.");
         return (-1);
     }
+    conf->publish = 0;
 
-    status = mosquitto_connect (conf->mosq, conf->host, conf->port,
-            /* keepalive = */ 10, /* clean session = */ 1);
-    if (status != MOSQ_ERR_SUCCESS)
+    conf->name = NULL;
+    status = cf_util_get_string (ci, &conf->name);
+    if (status != 0)
     {
-        char errbuf[1024];
-        ERROR ("mqtt plugin: mosquitto_connect failed: %s",
-                (status == MOSQ_ERR_ERRNO)
-                ? sstrerror (errno, errbuf, sizeof (errbuf))
-                : mosquitto_strerror (status));
-        free (conf);
+        mqtt_free (conf);
+        return (status);
+    }
+
+    conf->host = strdup (MQTT_DEFAULT_HOST);
+    conf->port = MQTT_DEFAULT_PORT;
+    conf->client_id = NULL;
+    conf->qos = 2;
+    conf->topic = strdup (MQTT_DEFAULT_TOPIC);
+    conf->clean_session = 1;
+
+    C_COMPLAIN_INIT (&conf->complaint_cantpublish);
+
+    for (i = 0; i < ci->children_num; i++)
+    {
+        oconfig_item_t *child = ci->children + i;
+        if (strcasecmp ("Host", child->key) == 0)
+            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)
+                ERROR ("mqtt plugin: Invalid port number.");
+            else
+                conf->port = tmp;
+        }
+        else if (strcasecmp ("ClientId", child->key) == 0)
+            cf_util_get_string (child, &conf->client_id);
+        else if (strcasecmp ("User", child->key) == 0)
+            cf_util_get_string (child, &conf->username);
+        else if (strcasecmp ("Password", child->key) == 0)
+            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))
+                ERROR ("mqtt plugin: Not a valid QoS setting.");
+            else
+                conf->qos = tmp;
+        }
+        else if (strcasecmp ("Topic", child->key) == 0)
+            cf_util_get_string (child, &conf->topic);
+        else if (strcasecmp ("CleanSession", child->key) == 0)
+            cf_util_get_boolean (child, &conf->clean_session);
+        else
+            ERROR ("mqtt plugin: Unknown config option: %s", child->key);
+    }
+
+    tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num);
+    if (tmp == NULL)
+    {
+        ERROR ("mqtt plugin: realloc failed.");
+        mqtt_free (conf);
         return (-1);
     }
+    subscribers = tmp;
+    subscribers[subscribers_num] = conf;
+    subscribers_num++;
 
-    DEBUG ("mqtt plugin: successfully connected to broker \"%s:%d\"",
-        conf->host, conf->port);
+    return (0);
+} /* mqtt_config_subscriber */
 
-    conf->connected = 1;
+/*
+ * <Plugin mqtt>
+ *   <Publish "name">
+ *     # ...
+ *   </Publish>
+ *   <Subscribe "name">
+ *     # ...
+ *   </Subscribe>
+ * </Plugin>
+ */
+static int mqtt_config (oconfig_item_t *ci)
+{
+    int i;
+
+    for (i = 0; i < ci->children_num; i++)
+    {
+        oconfig_item_t *child = ci->children + i;
 
-    plugin_register_write ("mqtt", mqtt_write, &user_data);
+        if (strcasecmp ("Publish", child->key) == 0)
+            mqtt_config_publisher (child);
+        else if (strcasecmp ("Subscribe", child->key) == 0)
+            mqtt_config_subscriber (child);
+        else
+            ERROR ("mqtt plugin: Unknown config option: %s", child->key);
+    }
 
     return (0);
-} /* mqtt_config */
+} /* int mqtt_config */
 
 static int mqtt_init (void)
 {
-    mosquitto_lib_init();
+    size_t i;
+
+    mosquitto_lib_init ();
+
+    for (i = 0; i < subscribers_num; i++)
+    {
+        int status;
+
+        if (subscribers[i]->loop)
+            continue;
+
+        status = plugin_thread_create (&subscribers[i]->thread,
+                /* attrs = */ NULL,
+                /* func  = */ subscribers_thread,
+                /* args  = */ subscribers[i]);
+        if (status != 0)
+        {
+            char errbuf[1024];
+            ERROR ("mqtt plugin: pthread_create failed: %s",
+                    sstrerror (errno, errbuf, sizeof (errbuf)));
+            continue;
+        }
+    }
 
     return (0);
 } /* mqtt_init */