src/mqtt.c

   1 /**
   2  * collectd - src/mqtt.c
   3  * Copyright (C) 2014       Marc Falzon
   4  * Copyright (C) 2014,2015  Florian octo Forster
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a
   7  * copy of this software and associated documentation files (the "Software"),
   8  * to deal in the Software without restriction, including without limitation
   9  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
  10  * and/or sell copies of the Software, and to permit persons to whom the
  11  * Software is furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  21  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  22  * DEALINGS IN THE SOFTWARE.
  23  *
  24  * Authors:
  25  *   Marc Falzon <marc at baha dot mu>
  26  *   Florian octo Forster <octo at collectd.org>
  27  *   Jan-Piet Mens <jpmens at gmail.com>
  28  **/
  29
  30 // Reference: http://mosquitto.org/api/files/mosquitto-h.html
  31
  32
  33 #include "collectd.h"
  34 #include "common.h"
  35 #include "plugin.h"
  36 #include "utils_cache.h"
  37 #include "utils_complain.h"
  38
  39 #include <pthread.h>
  40
  41 #include <mosquitto.h>
  42
  43 #define MQTT_MAX_TOPIC_SIZE         1024
  44 #define MQTT_MAX_MESSAGE_SIZE       MQTT_MAX_TOPIC_SIZE + 1024
  45 #define MQTT_DEFAULT_HOST           "localhost"
  46 #define MQTT_DEFAULT_PORT           1883
  47 #define MQTT_DEFAULT_TOPIC_PREFIX   "collectd"
  48 #define MQTT_DEFAULT_TOPIC          "collectd/#"
  49 #ifndef MQTT_KEEPALIVE
  50 # define MQTT_KEEPALIVE 60
  51 #endif
  52 #ifndef SSL_VERIFY_PEER
  53 # define SSL_VERIFY_PEER  1
  54 #endif
  55
  56
  57 /*
  58  * Data types
  59  */
  60 struct mqtt_client_conf
  61 {
  62     _Bool               publish;
  63     char               *name;
  64
  65     struct mosquitto   *mosq;
  66     _Bool               connected;
  67
  68     char               *host;
  69     int                 port;
  70     char               *client_id;
  71     char               *username;
  72     char               *password;
  73     int                 qos;
  74     char                *cacertificatefile;
  75     char                *certificatefile;
  76     char                *certificatekeyfile;
  77     char                *tlsprotocol;
  78     char                *ciphersuite;
  79
  80     /* For publishing */
  81     char               *topic_prefix;
  82     _Bool               store_rates;
  83     _Bool               retain;
  84
  85     /* For subscribing */
  86     pthread_t           thread;
  87     _Bool               loop;
  88     char               *topic;
  89     _Bool               clean_session;
  90
  91     c_complain_t        complaint_cantpublish;
  92     pthread_mutex_t     lock;
  93 };
  94 typedef struct mqtt_client_conf mqtt_client_conf_t;
  95
  96 static mqtt_client_conf_t **subscribers = NULL;
  97 static size_t subscribers_num = 0;
  98
  99 /*
 100  * Functions
 101  */
 102 #if LIBMOSQUITTO_MAJOR == 0
 103 static char const *mosquitto_strerror (int code)
 104 {
 105     switch (code)
 106     {
 107         case MOSQ_ERR_SUCCESS: return "MOSQ_ERR_SUCCESS";
 108         case MOSQ_ERR_NOMEM: return "MOSQ_ERR_NOMEM";
 109         case MOSQ_ERR_PROTOCOL: return "MOSQ_ERR_PROTOCOL";
 110         case MOSQ_ERR_INVAL: return "MOSQ_ERR_INVAL";
 111         case MOSQ_ERR_NO_CONN: return "MOSQ_ERR_NO_CONN";
 112         case MOSQ_ERR_CONN_REFUSED: return "MOSQ_ERR_CONN_REFUSED";
 113         case MOSQ_ERR_NOT_FOUND: return "MOSQ_ERR_NOT_FOUND";
 114         case MOSQ_ERR_CONN_LOST: return "MOSQ_ERR_CONN_LOST";
 115         case MOSQ_ERR_SSL: return "MOSQ_ERR_SSL";
 116         case MOSQ_ERR_PAYLOAD_SIZE: return "MOSQ_ERR_PAYLOAD_SIZE";
 117         case MOSQ_ERR_NOT_SUPPORTED: return "MOSQ_ERR_NOT_SUPPORTED";
 118         case MOSQ_ERR_AUTH: return "MOSQ_ERR_AUTH";
 119         case MOSQ_ERR_ACL_DENIED: return "MOSQ_ERR_ACL_DENIED";
 120         case MOSQ_ERR_UNKNOWN: return "MOSQ_ERR_UNKNOWN";
 121         case MOSQ_ERR_ERRNO: return "MOSQ_ERR_ERRNO";
 122     }
 123
 124     return "UNKNOWN ERROR CODE";
 125 }
 126 #else
 127 /* provided by libmosquitto */
 128 #endif
 129
 130 static void mqtt_free (mqtt_client_conf_t *conf)
 131 {
 132     if (conf == NULL)
 133         return;
 134
 135     if (conf->connected)
 136         (void) mosquitto_disconnect (conf->mosq);
 137     conf->connected = 0;
 138     (void) mosquitto_destroy (conf->mosq);
 139
 140     sfree (conf->host);
 141     sfree (conf->username);
 142     sfree (conf->password);
 143     sfree (conf->client_id);
 144     sfree (conf->topic_prefix);
 145     sfree (conf);
 146 }
 147
 148 static char *strip_prefix (char *topic)
 149 {
 150     size_t num;
 151     size_t i;
 152
 153     num = 0;
 154     for (i = 0; topic[i] != 0; i++)
 155         if (topic[i] == '/')
 156             num++;
 157
 158     if (num < 2)
 159         return (NULL);
 160
 161     while (num > 2)
 162     {
 163         char *tmp = strchr (topic, '/');
 164         if (tmp == NULL)
 165             return (NULL);
 166         topic = tmp + 1;
 167         num--;
 168     }
 169
 170     return (topic);
 171 }
 172
 173 static void on_message (
 174 #if LIBMOSQUITTO_MAJOR == 0
 175 #else
 176         __attribute__((unused)) struct mosquitto *m,
 177 #endif
 178         __attribute__((unused)) void *arg,
 179         const struct mosquitto_message *msg)
 180 {
 181     value_list_t vl = VALUE_LIST_INIT;
 182     data_set_t const *ds;
 183     char *topic;
 184     char *name;
 185     char *payload;
 186     int status;
 187
 188     if (msg->payloadlen <= 0) {
 189         DEBUG ("mqtt plugin: message has empty payload");
 190         return;
 191     }
 192
 193     topic = strdup (msg->topic);
 194     name = strip_prefix (topic);
 195
 196     status = parse_identifier_vl (name, &vl);
 197     if (status != 0)
 198     {
 199         ERROR ("mqtt plugin: Unable to parse topic \"%s\".", topic);
 200         sfree (topic);
 201         return;
 202     }
 203     sfree (topic);
 204
 205     ds = plugin_get_ds (vl.type);
 206     if (ds == NULL)
 207     {
 208         ERROR ("mqtt plugin: Unknown type: \"%s\".", vl.type);
 209         return;
 210     }
 211
 212     vl.values = calloc (ds->ds_num, sizeof (*vl.values));
 213     if (vl.values == NULL)
 214     {
 215         ERROR ("mqtt plugin: calloc failed.");
 216         return;
 217     }
 218     vl.values_len = ds->ds_num;
 219
 220     payload = malloc (msg->payloadlen+1);
 221     if (payload == NULL)
 222     {
 223         ERROR ("mqtt plugin: malloc for payload buffer failed.");
 224         sfree (vl.values);
 225         return;
 226     }
 227     memmove (payload, msg->payload, msg->payloadlen);
 228     payload[msg->payloadlen] = 0;
 229
 230     DEBUG ("mqtt plugin: payload = \"%s\"", payload);
 231     status = parse_values (payload, &vl, ds);
 232     if (status != 0)
 233     {
 234         ERROR ("mqtt plugin: Unable to parse payload \"%s\".", payload);
 235         sfree (payload);
 236         sfree (vl.values);
 237         return;
 238     }
 239     sfree (payload);
 240
 241     plugin_dispatch_values (&vl);
 242     sfree (vl.values);
 243 } /* void on_message */
 244
 245 /* must hold conf->lock when calling. */
 246 static int mqtt_reconnect (mqtt_client_conf_t *conf)
 247 {
 248     int status;
 249
 250     if (conf->connected)
 251         return (0);
 252
 253     status = mosquitto_reconnect (conf->mosq);
 254     if (status != MOSQ_ERR_SUCCESS)
 255     {
 256         char errbuf[1024];
 257         ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
 258                 (status == MOSQ_ERR_ERRNO)
 259                 ? sstrerror(errno, errbuf, sizeof (errbuf))
 260                 : mosquitto_strerror (status));
 261         return (-1);
 262     }
 263
 264     conf->connected = 1;
 265
 266     c_release (LOG_INFO,
 267             &conf->complaint_cantpublish,
 268             "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
 269             conf->host, conf->port);
 270
 271     return (0);
 272 } /* mqtt_reconnect */
 273
 274 /* must hold conf->lock when calling. */
 275 static int mqtt_connect (mqtt_client_conf_t *conf)
 276 {
 277     char const *client_id;
 278     int status;
 279
 280     if (conf->mosq != NULL)
 281         return mqtt_reconnect (conf);
 282
 283     if (conf->client_id)
 284         client_id = conf->client_id;
 285     else
 286         client_id = hostname_g;
 287
 288 #if LIBMOSQUITTO_MAJOR == 0
 289     conf->mosq = mosquitto_new (client_id, /* user data = */ conf);
 290 #else
 291     conf->mosq = mosquitto_new (client_id, conf->clean_session, /* user data = */ conf);
 292 #endif
 293     if (conf->mosq == NULL)
 294     {
 295         ERROR ("mqtt plugin: mosquitto_new failed");
 296         return (-1);
 297     }
 298
 299 #if LIBMOSQUITTO_MAJOR != 0
 300     if (conf->cacertificatefile) {
 301         status = mosquitto_tls_set(conf->mosq, conf->cacertificatefile, NULL,
 302                                    conf->certificatefile, conf->certificatekeyfile, /* pw_callback */NULL);
 303         if (status != MOSQ_ERR_SUCCESS) {
 304             ERROR ("mqtt plugin: cannot mosquitto_tls_set: %s", mosquitto_strerror(status));
 305             mosquitto_destroy (conf->mosq);
 306             conf->mosq = NULL;
 307             return (-1);
 308         }
 309
 310         status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER, conf->tlsprotocol, conf->ciphersuite);
 311         if (status != MOSQ_ERR_SUCCESS) {
 312             ERROR ("mqtt plugin: cannot mosquitto_tls_opts_set: %s", mosquitto_strerror(status));
 313             mosquitto_destroy (conf->mosq);
 314             conf->mosq = NULL;
 315             return (-1);
 316         }
 317
 318         status = mosquitto_tls_insecure_set(conf->mosq, false);
 319         if (status != MOSQ_ERR_SUCCESS) {
 320             ERROR ("mqtt plugin: cannot mosquitto_tls_insecure_set: %s", mosquitto_strerror(status));
 321             mosquitto_destroy (conf->mosq);
 322             conf->mosq = NULL;
 323             return (-1);
 324         }
 325     }
 326 #endif
 327
 328     if (conf->username && conf->password)
 329     {
 330         status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password);
 331         if (status != MOSQ_ERR_SUCCESS)
 332         {
 333             char errbuf[1024];
 334             ERROR ("mqtt plugin: mosquitto_username_pw_set failed: %s",
 335                     (status == MOSQ_ERR_ERRNO)
 336                     ? sstrerror (errno, errbuf, sizeof (errbuf))
 337                     : mosquitto_strerror (status));
 338
 339             mosquitto_destroy (conf->mosq);
 340             conf->mosq = NULL;
 341             return (-1);
 342         }
 343     }
 344
 345 #if LIBMOSQUITTO_MAJOR == 0
 346     status = mosquitto_connect (conf->mosq, conf->host, conf->port,
 347             /* keepalive = */ MQTT_KEEPALIVE, /* clean session = */ conf->clean_session);
 348 #else
 349     status = mosquitto_connect (conf->mosq, conf->host, conf->port, MQTT_KEEPALIVE);
 350 #endif
 351     if (status != MOSQ_ERR_SUCCESS)
 352     {
 353         char errbuf[1024];
 354         ERROR ("mqtt plugin: mosquitto_connect failed: %s",
 355                 (status == MOSQ_ERR_ERRNO)
 356                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 357                 : mosquitto_strerror (status));
 358
 359         mosquitto_destroy (conf->mosq);
 360         conf->mosq = NULL;
 361         return (-1);
 362     }
 363
 364     if (!conf->publish)
 365     {
 366         mosquitto_message_callback_set (conf->mosq, on_message);
 367
 368         status = mosquitto_subscribe (conf->mosq,
 369                 /* message_id = */ NULL,
 370                 conf->topic, conf->qos);
 371         if (status != MOSQ_ERR_SUCCESS)
 372         {
 373             ERROR ("mqtt plugin: Subscribing to \"%s\" failed: %s",
 374                     conf->topic, mosquitto_strerror (status));
 375
 376             mosquitto_disconnect (conf->mosq);
 377             mosquitto_destroy (conf->mosq);
 378             conf->mosq = NULL;
 379             return (-1);
 380         }
 381     }
 382
 383     conf->connected = 1;
 384     return (0);
 385 } /* mqtt_connect */
 386
 387 static void *subscribers_thread (void *arg)
 388 {
 389     mqtt_client_conf_t *conf = arg;
 390     int status;
 391
 392     conf->loop = 1;
 393
 394     while (conf->loop)
 395     {
 396         status = mqtt_connect (conf);
 397         if (status != 0)
 398         {
 399             sleep (1);
 400             continue;
 401         }
 402
 403         /* The documentation says "0" would map to the default (1000ms), but
 404          * that does not work on some versions. */
 405 #if LIBMOSQUITTO_MAJOR == 0
 406         status = mosquitto_loop (conf->mosq, /* timeout = */ 1000 /* ms */);
 407 #else
 408         status = mosquitto_loop (conf->mosq,
 409                 /* timeout[ms] = */ 1000,
 410                 /* max_packets = */  100);
 411 #endif
 412         if (status == MOSQ_ERR_CONN_LOST)
 413         {
 414             conf->connected = 0;
 415             continue;
 416         }
 417         else if (status != MOSQ_ERR_SUCCESS)
 418         {
 419             ERROR ("mqtt plugin: mosquitto_loop failed: %s",
 420                     mosquitto_strerror (status));
 421             mosquitto_destroy (conf->mosq);
 422             conf->mosq = NULL;
 423             conf->connected = 0;
 424             continue;
 425         }
 426
 427         DEBUG ("mqtt plugin: mosquitto_loop succeeded.");
 428     } /* while (conf->loop) */
 429
 430     pthread_exit (0);
 431 } /* void *subscribers_thread */
 432
 433 static int publish (mqtt_client_conf_t *conf, char const *topic,
 434     void const *payload, size_t payload_len)
 435 {
 436     int status;
 437
 438     pthread_mutex_lock (&conf->lock);
 439
 440     status = mqtt_connect (conf);
 441     if (status != 0) {
 442         pthread_mutex_unlock (&conf->lock);
 443         ERROR ("mqtt plugin: unable to reconnect to broker");
 444         return (status);
 445     }
 446
 447     status = mosquitto_publish(conf->mosq, /* message_id */ NULL, topic,
 448 #if LIBMOSQUITTO_MAJOR == 0
 449             (uint32_t) payload_len, payload,
 450 #else
 451             (int) payload_len, payload,
 452 #endif
 453             conf->qos, conf->retain);
 454     if (status != MOSQ_ERR_SUCCESS)
 455     {
 456         char errbuf[1024];
 457         c_complain (LOG_ERR,
 458             &conf->complaint_cantpublish,
 459             "mqtt plugin: mosquitto_publish failed: %s",
 460             (status == MOSQ_ERR_ERRNO)
 461             ? sstrerror(errno, errbuf, sizeof (errbuf))
 462             : mosquitto_strerror(status));
 463         /* Mark our connection "down" regardless of the error as a safety
 464          * measure; we will try to reconnect the next time we have to publish a
 465          * message */
 466         conf->connected = 0;
 467
 468         pthread_mutex_unlock (&conf->lock);
 469         return (-1);
 470     }
 471
 472     pthread_mutex_unlock (&conf->lock);
 473     return (0);
 474 } /* int publish */
 475
 476 static int format_topic (char *buf, size_t buf_len,
 477     data_set_t const *ds, value_list_t const *vl,
 478     mqtt_client_conf_t *conf)
 479 {
 480     char name[MQTT_MAX_TOPIC_SIZE];
 481     int status;
 482
 483     if ((conf->topic_prefix == NULL) || (conf->topic_prefix[0] == 0))
 484         return (FORMAT_VL (buf, buf_len, vl));
 485
 486     status = FORMAT_VL (name, sizeof (name), vl);
 487     if (status != 0)
 488         return (status);
 489
 490     status = ssnprintf (buf, buf_len, "%s/%s", conf->topic_prefix, name);
 491     if ((status < 0) || (((size_t) status) >= buf_len))
 492         return (ENOMEM);
 493
 494     return (0);
 495 } /* int format_topic */
 496
 497 static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
 498     user_data_t *user_data)
 499 {
 500     mqtt_client_conf_t *conf;
 501     char topic[MQTT_MAX_TOPIC_SIZE];
 502     char payload[MQTT_MAX_MESSAGE_SIZE];
 503     int status = 0;
 504
 505     if ((user_data == NULL) || (user_data->data == NULL))
 506         return (EINVAL);
 507     conf = user_data->data;
 508
 509     status = format_topic (topic, sizeof (topic), ds, vl, conf);
 510     if (status != 0)
 511     {
 512         ERROR ("mqtt plugin: format_topic failed with status %d.", status);
 513         return (status);
 514     }
 515
 516     status = format_values (payload, sizeof (payload),
 517             ds, vl, conf->store_rates);
 518     if (status != 0)
 519     {
 520         ERROR ("mqtt plugin: format_values failed with status %d.", status);
 521         return (status);
 522     }
 523
 524     status = publish (conf, topic, payload, strlen (payload) + 1);
 525     if (status != 0)
 526     {
 527         ERROR ("mqtt plugin: publish failed: %s", mosquitto_strerror (status));
 528         return (status);
 529     }
 530
 531     return (status);
 532 } /* mqtt_write */
 533
 534 /*
 535  * <Publish "name">
 536  *   Host "example.com"
 537  *   Port 1883
 538  *   ClientId "collectd"
 539  *   User "guest"
 540  *   Password "secret"
 541  *   Prefix "collectd"
 542  *   StoreRates true
 543  *   Retain false
 544  *   QoS 0
 545  *   CACert "ca.pem"                    Enables TLS if set
 546  *   CertificateFile "client-cert.pem"          optional
 547  *   CertificateKeyFile "client-key.pem"                optional
 548  *   TLSProtocol "tlsv1.2"              optional
 549  * </Publish>
 550  */
 551 static int mqtt_config_publisher (oconfig_item_t *ci)
 552 {
 553     mqtt_client_conf_t *conf;
 554     char cb_name[1024];
 555     user_data_t user_data;
 556     int status;
 557     int i;
 558
 559     conf = calloc (1, sizeof (*conf));
 560     if (conf == NULL)
 561     {
 562         ERROR ("mqtt plugin: calloc failed.");
 563         return (-1);
 564     }
 565     conf->publish = 1;
 566
 567     conf->name = NULL;
 568     status = cf_util_get_string (ci, &conf->name);
 569     if (status != 0)
 570     {
 571         mqtt_free (conf);
 572         return (status);
 573     }
 574
 575     conf->host = strdup (MQTT_DEFAULT_HOST);
 576     conf->port = MQTT_DEFAULT_PORT;
 577     conf->client_id = NULL;
 578     conf->qos = 0;
 579     conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
 580     conf->store_rates = 1;
 581
 582     status = pthread_mutex_init (&conf->lock, NULL);
 583     if (status != 0)
 584     {
 585       mqtt_free (conf);
 586       return (status);
 587     }
 588
 589     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 590
 591     for (i = 0; i < ci->children_num; i++)
 592     {
 593         oconfig_item_t *child = ci->children + i;
 594         if (strcasecmp ("Host", child->key) == 0)
 595             cf_util_get_string (child, &conf->host);
 596         else if (strcasecmp ("Port", child->key) == 0)
 597         {
 598             int tmp = cf_util_get_port_number (child);
 599             if (tmp < 0)
 600                 ERROR ("mqtt plugin: Invalid port number.");
 601             else
 602                 conf->port = tmp;
 603         }
 604         else if (strcasecmp ("ClientId", child->key) == 0)
 605             cf_util_get_string (child, &conf->client_id);
 606         else if (strcasecmp ("User", child->key) == 0)
 607             cf_util_get_string (child, &conf->username);
 608         else if (strcasecmp ("Password", child->key) == 0)
 609             cf_util_get_string (child, &conf->password);
 610         else if (strcasecmp ("QoS", child->key) == 0)
 611         {
 612             int tmp = -1;
 613             status = cf_util_get_int (child, &tmp);
 614             if ((status != 0) || (tmp < 0) || (tmp > 2))
 615                 ERROR ("mqtt plugin: Not a valid QoS setting.");
 616             else
 617                 conf->qos = tmp;
 618         }
 619         else if (strcasecmp ("Prefix", child->key) == 0)
 620             cf_util_get_string (child, &conf->topic_prefix);
 621         else if (strcasecmp ("StoreRates", child->key) == 0)
 622             cf_util_get_boolean (child, &conf->store_rates);
 623         else if (strcasecmp ("Retain", child->key) == 0)
 624             cf_util_get_boolean (child, &conf->retain);
 625         else if (strcasecmp ("CACert", child->key) == 0)
 626             cf_util_get_string (child, &conf->cacertificatefile);
 627         else if (strcasecmp ("CertificateFile", child->key) == 0)
 628             cf_util_get_string (child, &conf->certificatefile);
 629         else if (strcasecmp ("CertificateKeyFile", child->key) == 0)
 630             cf_util_get_string (child, &conf->certificatekeyfile);
 631         else if (strcasecmp ("TLSProtocol", child->key) == 0)
 632             cf_util_get_string (child, &conf->tlsprotocol);
 633         else if (strcasecmp ("CipherSuite", child->key) == 0)
 634             cf_util_get_string (child, &conf->ciphersuite);
 635         else
 636             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 637     }
 638
 639     ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name);
 640     memset (&user_data, 0, sizeof (user_data));
 641     user_data.data = conf;
 642
 643     plugin_register_write (cb_name, mqtt_write, &user_data);
 644     return (0);
 645 } /* mqtt_config_publisher */
 646
 647 /*
 648  * <Subscribe "name">
 649  *   Host "example.com"
 650  *   Port 1883
 651  *   ClientId "collectd"
 652  *   User "guest"
 653  *   Password "secret"
 654  *   Topic "collectd/#"
 655  * </Subscribe>
 656  */
 657 static int mqtt_config_subscriber (oconfig_item_t *ci)
 658 {
 659     mqtt_client_conf_t **tmp;
 660     mqtt_client_conf_t *conf;
 661     int status;
 662     int i;
 663
 664     conf = calloc (1, sizeof (*conf));
 665     if (conf == NULL)
 666     {
 667         ERROR ("mqtt plugin: calloc failed.");
 668         return (-1);
 669     }
 670     conf->publish = 0;
 671
 672     conf->name = NULL;
 673     status = cf_util_get_string (ci, &conf->name);
 674     if (status != 0)
 675     {
 676         mqtt_free (conf);
 677         return (status);
 678     }
 679
 680     conf->host = strdup (MQTT_DEFAULT_HOST);
 681     conf->port = MQTT_DEFAULT_PORT;
 682     conf->client_id = NULL;
 683     conf->qos = 2;
 684     conf->topic = strdup (MQTT_DEFAULT_TOPIC);
 685     conf->clean_session = 1;
 686
 687     status = pthread_mutex_init (&conf->lock, NULL);
 688     if (status != 0)
 689     {
 690       mqtt_free (conf);
 691       return (status);
 692     }
 693
 694     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 695
 696     for (i = 0; i < ci->children_num; i++)
 697     {
 698         oconfig_item_t *child = ci->children + i;
 699         if (strcasecmp ("Host", child->key) == 0)
 700             cf_util_get_string (child, &conf->host);
 701         else if (strcasecmp ("Port", child->key) == 0)
 702         {
 703             status = cf_util_get_port_number (child);
 704             if (status < 0)
 705                 ERROR ("mqtt plugin: Invalid port number.");
 706             else
 707                 conf->port = status;
 708         }
 709         else if (strcasecmp ("ClientId", child->key) == 0)
 710             cf_util_get_string (child, &conf->client_id);
 711         else if (strcasecmp ("User", child->key) == 0)
 712             cf_util_get_string (child, &conf->username);
 713         else if (strcasecmp ("Password", child->key) == 0)
 714             cf_util_get_string (child, &conf->password);
 715         else if (strcasecmp ("QoS", child->key) == 0)
 716         {
 717             int qos = -1;
 718             status = cf_util_get_int (child, &qos);
 719             if ((status != 0) || (qos < 0) || (qos > 2))
 720                 ERROR ("mqtt plugin: Not a valid QoS setting.");
 721             else
 722                 conf->qos = qos;
 723         }
 724         else if (strcasecmp ("Topic", child->key) == 0)
 725             cf_util_get_string (child, &conf->topic);
 726         else if (strcasecmp ("CleanSession", child->key) == 0)
 727             cf_util_get_boolean (child, &conf->clean_session);
 728         else
 729             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 730     }
 731
 732     tmp = realloc (subscribers, sizeof (*subscribers) * (subscribers_num + 1) );
 733     if (tmp == NULL)
 734     {
 735         ERROR ("mqtt plugin: realloc failed.");
 736         mqtt_free (conf);
 737         return (-1);
 738     }
 739     subscribers = tmp;
 740     subscribers[subscribers_num] = conf;
 741     subscribers_num++;
 742
 743     return (0);
 744 } /* mqtt_config_subscriber */
 745
 746 /*
 747  * <Plugin mqtt>
 748  *   <Publish "name">
 749  *     # ...
 750  *   </Publish>
 751  *   <Subscribe "name">
 752  *     # ...
 753  *   </Subscribe>
 754  * </Plugin>
 755  */
 756 static int mqtt_config (oconfig_item_t *ci)
 757 {
 758     int i;
 759
 760     for (i = 0; i < ci->children_num; i++)
 761     {
 762         oconfig_item_t *child = ci->children + i;
 763
 764         if (strcasecmp ("Publish", child->key) == 0)
 765             mqtt_config_publisher (child);
 766         else if (strcasecmp ("Subscribe", child->key) == 0)
 767             mqtt_config_subscriber (child);
 768         else
 769             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 770     }
 771
 772     return (0);
 773 } /* int mqtt_config */
 774
 775 static int mqtt_init (void)
 776 {
 777     size_t i;
 778
 779     mosquitto_lib_init ();
 780
 781     for (i = 0; i < subscribers_num; i++)
 782     {
 783         int status;
 784
 785         if (subscribers[i]->loop)
 786             continue;
 787
 788         status = plugin_thread_create (&subscribers[i]->thread,
 789                 /* attrs = */ NULL,
 790                 /* func  = */ subscribers_thread,
 791                 /* args  = */ subscribers[i]);
 792         if (status != 0)
 793         {
 794             char errbuf[1024];
 795             ERROR ("mqtt plugin: pthread_create failed: %s",
 796                     sstrerror (errno, errbuf, sizeof (errbuf)));
 797             continue;
 798         }
 799     }
 800
 801     return (0);
 802 } /* mqtt_init */
 803
 804 void module_register (void)
 805 {
 806     plugin_register_complex_config ("mqtt", mqtt_config);
 807     plugin_register_init ("mqtt", mqtt_init);
 808 } /* void module_register */
 809
 810 /* vim: set sw=4 sts=4 et fdm=marker : */