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