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