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             || (((uint8_t *) msg->payload)[msg->payloadlen - 1] != 0))
 190         return;
 191
 192     topic = strdup (msg->topic);
 193     name = strip_prefix (topic);
 194
 195     status = parse_identifier_vl (name, &vl);
 196     if (status != 0)
 197     {
 198         ERROR ("mqtt plugin: Unable to parse topic \"%s\".", topic);
 199         sfree (topic);
 200         return;
 201     }
 202     sfree (topic);
 203
 204     ds = plugin_get_ds (vl.type);
 205     if (ds == NULL)
 206     {
 207         ERROR ("mqtt plugin: Unknown type: \"%s\".", vl.type);
 208         return;
 209     }
 210
 211     vl.values = calloc (ds->ds_num, sizeof (*vl.values));
 212     if (vl.values == NULL)
 213     {
 214         ERROR ("mqtt plugin: calloc failed.");
 215         return;
 216     }
 217     vl.values_len = ds->ds_num;
 218
 219     payload = strdup ((void *) msg->payload);
 220     DEBUG ("mqtt plugin: payload = \"%s\"", payload);
 221     status = parse_values (payload, &vl, ds);
 222     if (status != 0)
 223     {
 224         ERROR ("mqtt plugin: Unable to parse payload \"%s\".", payload);
 225         sfree (payload);
 226         sfree (vl.values);
 227         return;
 228     }
 229     sfree (payload);
 230
 231     plugin_dispatch_values (&vl);
 232     sfree (vl.values);
 233 } /* void on_message */
 234
 235 /* must hold conf->lock when calling. */
 236 static int mqtt_reconnect (mqtt_client_conf_t *conf)
 237 {
 238     int status;
 239
 240     if (conf->connected)
 241         return (0);
 242
 243     status = mosquitto_reconnect (conf->mosq);
 244     if (status != MOSQ_ERR_SUCCESS)
 245     {
 246         char errbuf[1024];
 247         ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
 248                 (status == MOSQ_ERR_ERRNO)
 249                 ? sstrerror(errno, errbuf, sizeof (errbuf))
 250                 : mosquitto_strerror (status));
 251         return (-1);
 252     }
 253
 254     conf->connected = 1;
 255
 256     c_release (LOG_INFO,
 257             &conf->complaint_cantpublish,
 258             "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
 259             conf->host, conf->port);
 260
 261     return (0);
 262 } /* mqtt_reconnect */
 263
 264 /* must hold conf->lock when calling. */
 265 static int mqtt_connect (mqtt_client_conf_t *conf)
 266 {
 267     char const *client_id;
 268     int status;
 269
 270     if (conf->mosq != NULL)
 271         return mqtt_reconnect (conf);
 272
 273     if (conf->client_id)
 274         client_id = conf->client_id;
 275     else
 276         client_id = hostname_g;
 277
 278 #if LIBMOSQUITTO_MAJOR == 0
 279     conf->mosq = mosquitto_new (client_id, /* user data = */ conf);
 280 #else
 281     conf->mosq = mosquitto_new (client_id, conf->clean_session, /* user data = */ conf);
 282 #endif
 283     if (conf->mosq == NULL)
 284     {
 285         ERROR ("mqtt plugin: mosquitto_new failed");
 286         return (-1);
 287     }
 288
 289 #if LIBMOSQUITTO_MAJOR != 0
 290     if (conf->cacertificatefile) {
 291         status = mosquitto_tls_set(conf->mosq, conf->cacertificatefile, NULL,
 292                                    conf->certificatefile, conf->certificatekeyfile, /* pw_callback */NULL);
 293         if (status != MOSQ_ERR_SUCCESS) {
 294             ERROR ("mqtt plugin: cannot mosquitto_tls_set: %s", mosquitto_strerror(status));
 295             mosquitto_destroy (conf->mosq);
 296             conf->mosq = NULL;
 297             return (-1);
 298         }
 299
 300         status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER, conf->tlsprotocol, conf->ciphersuite);
 301         if (status != MOSQ_ERR_SUCCESS) {
 302             ERROR ("mqtt plugin: cannot mosquitto_tls_opts_set: %s", mosquitto_strerror(status));
 303             mosquitto_destroy (conf->mosq);
 304             conf->mosq = NULL;
 305             return (-1);
 306         }
 307
 308         status = mosquitto_tls_insecure_set(conf->mosq, false);
 309         if (status != MOSQ_ERR_SUCCESS) {
 310             ERROR ("mqtt plugin: cannot mosquitto_tls_insecure_set: %s", mosquitto_strerror(status));
 311             mosquitto_destroy (conf->mosq);
 312             conf->mosq = NULL;
 313             return (-1);
 314         }
 315     }
 316 #endif
 317
 318     if (conf->username && conf->password)
 319     {
 320         status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password);
 321         if (status != MOSQ_ERR_SUCCESS)
 322         {
 323             char errbuf[1024];
 324             ERROR ("mqtt plugin: mosquitto_username_pw_set failed: %s",
 325                     (status == MOSQ_ERR_ERRNO)
 326                     ? sstrerror (errno, errbuf, sizeof (errbuf))
 327                     : mosquitto_strerror (status));
 328
 329             mosquitto_destroy (conf->mosq);
 330             conf->mosq = NULL;
 331             return (-1);
 332         }
 333     }
 334
 335 #if LIBMOSQUITTO_MAJOR == 0
 336     status = mosquitto_connect (conf->mosq, conf->host, conf->port,
 337             /* keepalive = */ MQTT_KEEPALIVE, /* clean session = */ conf->clean_session);
 338 #else
 339     status = mosquitto_connect (conf->mosq, conf->host, conf->port, MQTT_KEEPALIVE);
 340 #endif
 341     if (status != MOSQ_ERR_SUCCESS)
 342     {
 343         char errbuf[1024];
 344         ERROR ("mqtt plugin: mosquitto_connect failed: %s",
 345                 (status == MOSQ_ERR_ERRNO)
 346                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 347                 : mosquitto_strerror (status));
 348
 349         mosquitto_destroy (conf->mosq);
 350         conf->mosq = NULL;
 351         return (-1);
 352     }
 353
 354     if (!conf->publish)
 355     {
 356         mosquitto_message_callback_set (conf->mosq, on_message);
 357
 358         status = mosquitto_subscribe (conf->mosq,
 359                 /* message_id = */ NULL,
 360                 conf->topic, conf->qos);
 361         if (status != MOSQ_ERR_SUCCESS)
 362         {
 363             ERROR ("mqtt plugin: Subscribing to \"%s\" failed: %s",
 364                     conf->topic, mosquitto_strerror (status));
 365
 366             mosquitto_disconnect (conf->mosq);
 367             mosquitto_destroy (conf->mosq);
 368             conf->mosq = NULL;
 369             return (-1);
 370         }
 371     }
 372
 373     conf->connected = 1;
 374     return (0);
 375 } /* mqtt_connect */
 376
 377 static void *subscribers_thread (void *arg)
 378 {
 379     mqtt_client_conf_t *conf = arg;
 380     int status;
 381
 382     conf->loop = 1;
 383
 384     while (conf->loop)
 385     {
 386         status = mqtt_connect (conf);
 387         if (status != 0)
 388         {
 389             sleep (1);
 390             continue;
 391         }
 392
 393         /* The documentation says "0" would map to the default (1000ms), but
 394          * that does not work on some versions. */
 395 #if LIBMOSQUITTO_MAJOR == 0
 396         status = mosquitto_loop (conf->mosq, /* timeout = */ 1000 /* ms */);
 397 #else
 398         status = mosquitto_loop (conf->mosq,
 399                 /* timeout[ms] = */ 1000,
 400                 /* max_packets = */  100);
 401 #endif
 402         if (status == MOSQ_ERR_CONN_LOST)
 403         {
 404             conf->connected = 0;
 405             continue;
 406         }
 407         else if (status != MOSQ_ERR_SUCCESS)
 408         {
 409             ERROR ("mqtt plugin: mosquitto_loop failed: %s",
 410                     mosquitto_strerror (status));
 411             mosquitto_destroy (conf->mosq);
 412             conf->mosq = NULL;
 413             conf->connected = 0;
 414             continue;
 415         }
 416
 417         DEBUG ("mqtt plugin: mosquitto_loop succeeded.");
 418     } /* while (conf->loop) */
 419
 420     pthread_exit (0);
 421 } /* void *subscribers_thread */
 422
 423 static int publish (mqtt_client_conf_t *conf, char const *topic,
 424     void const *payload, size_t payload_len)
 425 {
 426     int status;
 427
 428     pthread_mutex_lock (&conf->lock);
 429
 430     status = mqtt_connect (conf);
 431     if (status != 0) {
 432         pthread_mutex_unlock (&conf->lock);
 433         ERROR ("mqtt plugin: unable to reconnect to broker");
 434         return (status);
 435     }
 436
 437     status = mosquitto_publish(conf->mosq, /* message_id */ NULL, topic,
 438 #if LIBMOSQUITTO_MAJOR == 0
 439             (uint32_t) payload_len, payload,
 440 #else
 441             (int) payload_len, payload,
 442 #endif
 443             conf->qos, conf->retain);
 444     if (status != MOSQ_ERR_SUCCESS)
 445     {
 446         char errbuf[1024];
 447         c_complain (LOG_ERR,
 448             &conf->complaint_cantpublish,
 449             "mqtt plugin: mosquitto_publish failed: %s",
 450             (status == MOSQ_ERR_ERRNO)
 451             ? sstrerror(errno, errbuf, sizeof (errbuf))
 452             : mosquitto_strerror(status));
 453         /* Mark our connection "down" regardless of the error as a safety
 454          * measure; we will try to reconnect the next time we have to publish a
 455          * message */
 456         conf->connected = 0;
 457
 458         pthread_mutex_unlock (&conf->lock);
 459         return (-1);
 460     }
 461
 462     pthread_mutex_unlock (&conf->lock);
 463     return (0);
 464 } /* int publish */
 465
 466 static int format_topic (char *buf, size_t buf_len,
 467     data_set_t const *ds, value_list_t const *vl,
 468     mqtt_client_conf_t *conf)
 469 {
 470     char name[MQTT_MAX_TOPIC_SIZE];
 471     int status;
 472
 473     if ((conf->topic_prefix == NULL) || (conf->topic_prefix[0] == 0))
 474         return (FORMAT_VL (buf, buf_len, vl));
 475
 476     status = FORMAT_VL (name, sizeof (name), vl);
 477     if (status != 0)
 478         return (status);
 479
 480     status = ssnprintf (buf, buf_len, "%s/%s", conf->topic_prefix, name);
 481     if ((status < 0) || (((size_t) status) >= buf_len))
 482         return (ENOMEM);
 483
 484     return (0);
 485 } /* int format_topic */
 486
 487 static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
 488     user_data_t *user_data)
 489 {
 490     mqtt_client_conf_t *conf;
 491     char topic[MQTT_MAX_TOPIC_SIZE];
 492     char payload[MQTT_MAX_MESSAGE_SIZE];
 493     int status = 0;
 494
 495     if ((user_data == NULL) || (user_data->data == NULL))
 496         return (EINVAL);
 497     conf = user_data->data;
 498
 499     status = format_topic (topic, sizeof (topic), ds, vl, conf);
 500     if (status != 0)
 501     {
 502         ERROR ("mqtt plugin: format_topic failed with status %d.", status);
 503         return (status);
 504     }
 505
 506     status = format_values (payload, sizeof (payload),
 507             ds, vl, conf->store_rates);
 508     if (status != 0)
 509     {
 510         ERROR ("mqtt plugin: format_values failed with status %d.", status);
 511         return (status);
 512     }
 513
 514     status = publish (conf, topic, payload, strlen (payload) + 1);
 515     if (status != 0)
 516     {
 517         ERROR ("mqtt plugin: publish failed: %s", mosquitto_strerror (status));
 518         return (status);
 519     }
 520
 521     return (status);
 522 } /* mqtt_write */
 523
 524 /*
 525  * <Publish "name">
 526  *   Host "example.com"
 527  *   Port 1883
 528  *   ClientId "collectd"
 529  *   User "guest"
 530  *   Password "secret"
 531  *   Prefix "collectd"
 532  *   StoreRates true
 533  *   Retain false
 534  *   QoS 0
 535  *   CACertificateFile "ca.pem"                 Enables TLS if set
 536  *   CertificateFile "client-cert.pem"          optional
 537  *   CertificateKeyFile "client-key.pem"                optional
 538  *   TLSprotocol "tlsv1.2"              optional
 539  * </Publish>
 540  */
 541 static int mqtt_config_publisher (oconfig_item_t *ci)
 542 {
 543     mqtt_client_conf_t *conf;
 544     char cb_name[1024];
 545     user_data_t user_data;
 546     int status;
 547     int i;
 548
 549     conf = calloc (1, sizeof (*conf));
 550     if (conf == NULL)
 551     {
 552         ERROR ("mqtt plugin: malloc failed.");
 553         return (-1);
 554     }
 555     conf->publish = 1;
 556
 557     conf->name = NULL;
 558     status = cf_util_get_string (ci, &conf->name);
 559     if (status != 0)
 560     {
 561         mqtt_free (conf);
 562         return (status);
 563     }
 564
 565     conf->host = strdup (MQTT_DEFAULT_HOST);
 566     conf->port = MQTT_DEFAULT_PORT;
 567     conf->client_id = NULL;
 568     conf->qos = 0;
 569     conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
 570     conf->store_rates = 1;
 571
 572     status = pthread_mutex_init (&conf->lock, NULL);
 573     if (status != 0)
 574     {
 575       mqtt_free (conf);
 576       return (status);
 577     }
 578
 579     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 580
 581     for (i = 0; i < ci->children_num; i++)
 582     {
 583         oconfig_item_t *child = ci->children + i;
 584         if (strcasecmp ("Host", child->key) == 0)
 585             cf_util_get_string (child, &conf->host);
 586         else if (strcasecmp ("Port", child->key) == 0)
 587         {
 588             int tmp = cf_util_get_port_number (child);
 589             if (tmp < 0)
 590                 ERROR ("mqtt plugin: Invalid port number.");
 591             else
 592                 conf->port = tmp;
 593         }
 594         else if (strcasecmp ("ClientId", child->key) == 0)
 595             cf_util_get_string (child, &conf->client_id);
 596         else if (strcasecmp ("User", child->key) == 0)
 597             cf_util_get_string (child, &conf->username);
 598         else if (strcasecmp ("Password", child->key) == 0)
 599             cf_util_get_string (child, &conf->password);
 600         else if (strcasecmp ("QoS", child->key) == 0)
 601         {
 602             int tmp = -1;
 603             status = cf_util_get_int (child, &tmp);
 604             if ((status != 0) || (tmp < 0) || (tmp > 2))
 605                 ERROR ("mqtt plugin: Not a valid QoS setting.");
 606             else
 607                 conf->qos = tmp;
 608         }
 609         else if (strcasecmp ("Prefix", child->key) == 0)
 610             cf_util_get_string (child, &conf->topic_prefix);
 611         else if (strcasecmp ("StoreRates", child->key) == 0)
 612             cf_util_get_boolean (child, &conf->store_rates);
 613         else if (strcasecmp ("Retain", child->key) == 0)
 614             cf_util_get_boolean (child, &conf->retain);
 615         else if (strcasecmp ("CACertificateFile", child->key) == 0)
 616             cf_util_get_string (child, &conf->cacertificatefile);
 617         else if (strcasecmp ("CertificateFile", child->key) == 0)
 618             cf_util_get_string (child, &conf->certificatefile);
 619         else if (strcasecmp ("CertificateKeyFile", child->key) == 0)
 620             cf_util_get_string (child, &conf->certificatekeyfile);
 621         else if (strcasecmp ("TLSprotocol", child->key) == 0)
 622             cf_util_get_string (child, &conf->tlsprotocol);
 623         else if (strcasecmp ("CipherSuite", child->key) == 0)
 624             cf_util_get_string (child, &conf->ciphersuite);
 625         else
 626             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 627     }
 628
 629     ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name);
 630     memset (&user_data, 0, sizeof (user_data));
 631     user_data.data = conf;
 632
 633     plugin_register_write (cb_name, mqtt_write, &user_data);
 634     return (0);
 635 } /* mqtt_config_publisher */
 636
 637 /*
 638  * <Subscribe "name">
 639  *   Host "example.com"
 640  *   Port 1883
 641  *   ClientId "collectd"
 642  *   User "guest"
 643  *   Password "secret"
 644  *   Topic "collectd/#"
 645  * </Publish>
 646  */
 647 static int mqtt_config_subscriber (oconfig_item_t *ci)
 648 {
 649     mqtt_client_conf_t **tmp;
 650     mqtt_client_conf_t *conf;
 651     int status;
 652     int i;
 653
 654     conf = calloc (1, sizeof (*conf));
 655     if (conf == NULL)
 656     {
 657         ERROR ("mqtt plugin: malloc failed.");
 658         return (-1);
 659     }
 660     conf->publish = 0;
 661
 662     conf->name = NULL;
 663     status = cf_util_get_string (ci, &conf->name);
 664     if (status != 0)
 665     {
 666         mqtt_free (conf);
 667         return (status);
 668     }
 669
 670     conf->host = strdup (MQTT_DEFAULT_HOST);
 671     conf->port = MQTT_DEFAULT_PORT;
 672     conf->client_id = NULL;
 673     conf->qos = 2;
 674     conf->topic = strdup (MQTT_DEFAULT_TOPIC);
 675     conf->clean_session = 1;
 676
 677     status = pthread_mutex_init (&conf->lock, NULL);
 678     if (status != 0)
 679     {
 680       mqtt_free (conf);
 681       return (status);
 682     }
 683
 684     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 685
 686     for (i = 0; i < ci->children_num; i++)
 687     {
 688         oconfig_item_t *child = ci->children + i;
 689         if (strcasecmp ("Host", child->key) == 0)
 690             cf_util_get_string (child, &conf->host);
 691         else if (strcasecmp ("Port", child->key) == 0)
 692         {
 693             int tmp = cf_util_get_port_number (child);
 694             if (tmp < 0)
 695                 ERROR ("mqtt plugin: Invalid port number.");
 696             else
 697                 conf->port = tmp;
 698         }
 699         else if (strcasecmp ("ClientId", child->key) == 0)
 700             cf_util_get_string (child, &conf->client_id);
 701         else if (strcasecmp ("User", child->key) == 0)
 702             cf_util_get_string (child, &conf->username);
 703         else if (strcasecmp ("Password", child->key) == 0)
 704             cf_util_get_string (child, &conf->password);
 705         else if (strcasecmp ("QoS", child->key) == 0)
 706         {
 707             int tmp = -1;
 708             status = cf_util_get_int (child, &tmp);
 709             if ((status != 0) || (tmp < 0) || (tmp > 2))
 710                 ERROR ("mqtt plugin: Not a valid QoS setting.");
 711             else
 712                 conf->qos = tmp;
 713         }
 714         else if (strcasecmp ("Topic", child->key) == 0)
 715             cf_util_get_string (child, &conf->topic);
 716         else if (strcasecmp ("CleanSession", child->key) == 0)
 717             cf_util_get_boolean (child, &conf->clean_session);
 718         else
 719             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 720     }
 721
 722     tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num);
 723     if (tmp == NULL)
 724     {
 725         ERROR ("mqtt plugin: realloc failed.");
 726         mqtt_free (conf);
 727         return (-1);
 728     }
 729     subscribers = tmp;
 730     subscribers[subscribers_num] = conf;
 731     subscribers_num++;
 732
 733     return (0);
 734 } /* mqtt_config_subscriber */
 735
 736 /*
 737  * <Plugin mqtt>
 738  *   <Publish "name">
 739  *     # ...
 740  *   </Publish>
 741  *   <Subscribe "name">
 742  *     # ...
 743  *   </Subscribe>
 744  * </Plugin>
 745  */
 746 static int mqtt_config (oconfig_item_t *ci)
 747 {
 748     int i;
 749
 750     for (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     size_t i;
 768
 769     mosquitto_lib_init ();
 770
 771     for (i = 0; i < subscribers_num; i++)
 772     {
 773         int status;
 774
 775         if (subscribers[i]->loop)
 776             continue;
 777
 778         status = plugin_thread_create (&subscribers[i]->thread,
 779                 /* attrs = */ NULL,
 780                 /* func  = */ subscribers_thread,
 781                 /* args  = */ subscribers[i]);
 782         if (status != 0)
 783         {
 784             char errbuf[1024];
 785             ERROR ("mqtt plugin: pthread_create failed: %s",
 786                     sstrerror (errno, errbuf, sizeof (errbuf)));
 787             continue;
 788         }
 789     }
 790
 791     return (0);
 792 } /* mqtt_init */
 793
 794 void module_register (void)
 795 {
 796     plugin_register_complex_config ("mqtt", mqtt_config);
 797     plugin_register_init ("mqtt", mqtt_init);
 798 } /* void module_register */
 799
 800 /* vim: set sw=4 sts=4 et fdm=marker : */