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