X-Git-Url: https://git.verplant.org/?a=blobdiff_plain;f=src%2Fmqtt.c;h=7a7ba59dfb57b06d850ebb5f13619c5291b5f8c2;hb=1f2b236c9aa0aec7c5ca9d69cc36d9f7742578e3;hp=fffa8fadd3115bff298ba97c23f16e8030847bf9;hpb=5a4743253477982a890bf1aa2fba67f6275fd2df;p=collectd.git

diff --git a/src/mqtt.c b/src/mqtt.c
index fffa8fad..7a7ba59d 100644
--- a/src/mqtt.c
+++ b/src/mqtt.c
@@ -38,25 +38,49 @@
 #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/#"
 
 /*
  * Data types
  */
 struct mqtt_client_conf
 {
-    struct mosquitto    *mosq;
-    bool                connected;
-    char                *host;
+    _Bool               publish;
+    char               *name;
+
+    struct mosquitto   *mosq;
+    _Bool               connected;
+
+    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 +105,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;
 
@@ -95,31 +215,147 @@ static int mqtt_reconnect_broker (mqtt_client_conf_t *conf)
     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 ?
-                strerror(errno) : mosquitto_strerror (status)));
+                (status == MOSQ_ERR_ERRNO)
+                ? sstrerror(errno, errbuf, sizeof (errbuf))
+                : mosquitto_strerror (status));
         return (-1);
     }
 
-    conf->connected = true;
+    conf->connected = 1;
 
     c_release (LOG_INFO,
-        &conf->complaint_cantpublish,
-        "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
-        conf->host, conf->port);
+            &conf->complaint_cantpublish,
+            "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
+            conf->host, conf->port);
+
+    return (0);
+} /* 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 = */ 10, /* 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_reconnect_broker */
+} /* 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");
@@ -130,8 +366,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];
@@ -144,7 +380,7 @@ static int publish (mqtt_client_conf_t *conf, char const *topic,
         /* 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 = false;
+        conf->connected = 0;
 
         pthread_mutex_unlock (&conf->lock);
         return (-1);
@@ -182,27 +418,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));
@@ -212,11 +448,25 @@ static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
     return (status);
 } /* mqtt_write */
 
-static int mqtt_config (oconfig_item_t *ci)
+/*
+ * <Publish "name">
+ *   Host "example.com"
+ *   Port 1883
+ *   ClientId "collectd"
+ *   User "guest"
+ *   Password "secret"
+ *   Prefix "collectd"
+ *   StoreRates true
+ *   Retain false
+ *   QoS 0
+ * </Publish>
+ */
+static int mqtt_config_publisher (oconfig_item_t *ci)
 {
     mqtt_client_conf_t *conf;
     user_data_t user_data;
     int status;
+    int i;
 
     conf = calloc (1, sizeof (*conf));
     if (conf == NULL)
@@ -224,51 +474,212 @@ static int mqtt_config (oconfig_item_t *ci)
         ERROR ("mqtt plugin: malloc failed.");
         return (-1);
     }
+    conf->publish = 1;
 
-    conf->connected = false;
-    conf->host = MQTT_DEFAULT_HOST;
+    conf->name = NULL;
+    status = cf_util_get_string (ci, &conf->name);
+    if (status != 0)
+    {
+        mqtt_free (conf);
+        return (status);
+    }
+
+    conf->host = strdup (MQTT_DEFAULT_HOST);
     conf->port = MQTT_DEFAULT_PORT;
-    conf->client_id = MQTT_DEFAULT_CLIENT_ID;
-    conf->topic_prefix = MQTT_DEFAULT_TOPIC_PREFIX;
+    conf->client_id = NULL;
+    conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
+
     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 ("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);
+    }
+
     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 ("mqtt", 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->topic = strdup (MQTT_DEFAULT_TOPIC);
+
+    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 = true;
+/*
+ * <Plugin mqtt>
+ *   <Publish "name">
+ *     # ...
+ *   </Publish>
+ *   <Subscribe "name">
+ *     # ...
+ *   </Subscribe>
+ * </Plugin>
+ */
+static int mqtt_config (oconfig_item_t *ci)
+{
+    int i;
 
-    plugin_register_write ("mqtt", mqtt_write, &user_data);
+    for (i = 0; i < ci->children_num; i++)
+    {
+        oconfig_item_t *child = ci->children + i;
+
+        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 */