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