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