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     user_data_t user_data;
 477     int status;
 478     int i;
 479
 480     conf = calloc (1, sizeof (*conf));
 481     if (conf == NULL)
 482     {
 483         ERROR ("mqtt plugin: malloc failed.");
 484         return (-1);
 485     }
 486     conf->publish = 1;
 487
 488     conf->name = NULL;
 489     status = cf_util_get_string (ci, &conf->name);
 490     if (status != 0)
 491     {
 492         mqtt_free (conf);
 493         return (status);
 494     }
 495
 496     conf->host = strdup (MQTT_DEFAULT_HOST);
 497     conf->port = MQTT_DEFAULT_PORT;
 498     conf->client_id = NULL;
 499     conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
 500
 501     C_COMPLAIN_INIT (&conf->complaint_cantpublish);
 502
 503     for (i = 0; i < ci->children_num; i++)
 504     {
 505         oconfig_item_t *child = ci->children + i;
 506         if (strcasecmp ("Host", child->key) == 0)
 507             cf_util_get_string (child, &conf->host);
 508         else if (strcasecmp ("Port", child->key) == 0)
 509         {
 510             int tmp = cf_util_get_port_number (child);
 511             if (tmp < 0)
 512                 ERROR ("mqtt plugin: Invalid port number.");
 513             else
 514                 conf->port = tmp;
 515         }
 516         else if (strcasecmp ("ClientId", child->key) == 0)
 517             cf_util_get_string (child, &conf->client_id);
 518         else if (strcasecmp ("User", child->key) == 0)
 519             cf_util_get_string (child, &conf->username);
 520         else if (strcasecmp ("Password", child->key) == 0)
 521             cf_util_get_string (child, &conf->password);
 522         else if (strcasecmp ("QoS", child->key) == 0)
 523         {
 524             int tmp = -1;
 525             status = cf_util_get_int (child, &tmp);
 526             if ((status != 0) || (tmp < 0) || (tmp > 2))
 527                 ERROR ("mqtt plugin: Not a valid QoS setting.");
 528             else
 529                 conf->qos = tmp;
 530         }
 531         else if (strcasecmp ("Prefix", child->key) == 0)
 532             cf_util_get_string (child, &conf->topic_prefix);
 533         else if (strcasecmp ("StoreRates", child->key) == 0)
 534             cf_util_get_boolean (child, &conf->store_rates);
 535         else if (strcasecmp ("Retain", child->key) == 0)
 536             cf_util_get_boolean (child, &conf->retain);
 537         else
 538             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 539     }
 540
 541     memset (&user_data, 0, sizeof (user_data));
 542     user_data.data = conf;
 543
 544     plugin_register_write ("mqtt", mqtt_write, &user_data);
 545     return (0);
 546 } /* mqtt_config_publisher */
 547
 548 /*
 549  * <Subscribe "name">
 550  *   Host "example.com"
 551  *   Port 1883
 552  *   ClientId "collectd"
 553  *   User "guest"
 554  *   Password "secret"
 555  *   Topic "collectd/#"
 556  * </Publish>
 557  */
 558 static int mqtt_config_subscriber (oconfig_item_t *ci)
 559 {
 560     mqtt_client_conf_t **tmp;
 561     mqtt_client_conf_t *conf;
 562     int status;
 563     int i;
 564
 565     conf = calloc (1, sizeof (*conf));
 566     if (conf == NULL)
 567     {
 568         ERROR ("mqtt plugin: malloc failed.");
 569         return (-1);
 570     }
 571     conf->publish = 0;
 572
 573     conf->name = NULL;
 574     status = cf_util_get_string (ci, &conf->name);
 575     if (status != 0)
 576     {
 577         mqtt_free (conf);
 578         return (status);
 579     }
 580
 581     conf->host = strdup (MQTT_DEFAULT_HOST);
 582     conf->port = MQTT_DEFAULT_PORT;
 583     conf->client_id = NULL;
 584     conf->topic = strdup (MQTT_DEFAULT_TOPIC);
 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 ("Topic", child->key) == 0)
 617             cf_util_get_string (child, &conf->topic);
 618         else if (strcasecmp ("CleanSession", child->key) == 0)
 619             cf_util_get_boolean (child, &conf->clean_session);
 620         else
 621             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 622     }
 623
 624     tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num);
 625     if (tmp == NULL)
 626     {
 627         ERROR ("mqtt plugin: realloc failed.");
 628         mqtt_free (conf);
 629         return (-1);
 630     }
 631     subscribers = tmp;
 632     subscribers[subscribers_num] = conf;
 633     subscribers_num++;
 634
 635     return (0);
 636 } /* mqtt_config_subscriber */
 637
 638 /*
 639  * <Plugin mqtt>
 640  *   <Publish "name">
 641  *     # ...
 642  *   </Publish>
 643  *   <Subscribe "name">
 644  *     # ...
 645  *   </Subscribe>
 646  * </Plugin>
 647  */
 648 static int mqtt_config (oconfig_item_t *ci)
 649 {
 650     int i;
 651
 652     for (i = 0; i < ci->children_num; i++)
 653     {
 654         oconfig_item_t *child = ci->children + i;
 655
 656         if (strcasecmp ("Publish", child->key) == 0)
 657             mqtt_config_publisher (child);
 658         else if (strcasecmp ("Subscribe", child->key) == 0)
 659             mqtt_config_subscriber (child);
 660         else
 661             ERROR ("mqtt plugin: Unknown config option: %s", child->key);
 662     }
 663
 664     return (0);
 665 } /* int mqtt_config */
 666
 667 static int mqtt_init (void)
 668 {
 669     size_t i;
 670
 671     mosquitto_lib_init ();
 672
 673     for (i = 0; i < subscribers_num; i++)
 674     {
 675         int status;
 676
 677         if (subscribers[i]->loop)
 678             continue;
 679
 680         status = plugin_thread_create (&subscribers[i]->thread,
 681                 /* attrs = */ NULL,
 682                 /* func  = */ subscribers_thread,
 683                 /* args  = */ subscribers[i]);
 684         if (status != 0)
 685         {
 686             char errbuf[1024];
 687             ERROR ("mqtt plugin: pthread_create failed: %s",
 688                     sstrerror (errno, errbuf, sizeof (errbuf)));
 689             continue;
 690         }
 691     }
 692
 693     return (0);
 694 } /* mqtt_init */
 695
 696 void module_register (void)
 697 {
 698     plugin_register_complex_config ("mqtt", mqtt_config);
 699     plugin_register_init ("mqtt", mqtt_init);
 700 } /* void module_register */
 701
 702 /* vim: set sw=4 sts=4 et fdm=marker : */