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