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