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 DEBUG ("mqtt plugin: message has empty payload");
190 return;
191 }
193 topic = strdup (msg->topic);
194 name = strip_prefix (topic);
196 status = parse_identifier_vl (name, &vl);
197 if (status != 0)
198 {
199 ERROR ("mqtt plugin: Unable to parse topic \"%s\".", topic);
200 sfree (topic);
201 return;
202 }
203 sfree (topic);
205 ds = plugin_get_ds (vl.type);
206 if (ds == NULL)
207 {
208 ERROR ("mqtt plugin: Unknown type: \"%s\".", vl.type);
209 return;
210 }
212 vl.values = calloc (ds->ds_num, sizeof (*vl.values));
213 if (vl.values == NULL)
214 {
215 ERROR ("mqtt plugin: calloc failed.");
216 return;
217 }
218 vl.values_len = ds->ds_num;
220 payload = malloc (msg->payloadlen+1);
221 if (payload == NULL)
222 {
223 ERROR ("mqtt plugin: malloc for payload buffer failed.");
224 return;
225 }
226 memmove (payload, msg->payload, msg->payloadlen);
227 payload[msg->payloadlen] = 0;
229 DEBUG ("mqtt plugin: payload = \"%s\"", payload);
230 status = parse_values (payload, &vl, ds);
231 if (status != 0)
232 {
233 ERROR ("mqtt plugin: Unable to parse payload \"%s\".", payload);
234 sfree (payload);
235 sfree (vl.values);
236 return;
237 }
238 sfree (payload);
240 plugin_dispatch_values (&vl);
241 sfree (vl.values);
242 } /* void on_message */
244 /* must hold conf->lock when calling. */
245 static int mqtt_reconnect (mqtt_client_conf_t *conf)
246 {
247 int status;
249 if (conf->connected)
250 return (0);
252 status = mosquitto_reconnect (conf->mosq);
253 if (status != MOSQ_ERR_SUCCESS)
254 {
255 char errbuf[1024];
256 ERROR ("mqtt_connect_broker: mosquitto_connect failed: %s",
257 (status == MOSQ_ERR_ERRNO)
258 ? sstrerror(errno, errbuf, sizeof (errbuf))
259 : mosquitto_strerror (status));
260 return (-1);
261 }
263 conf->connected = 1;
265 c_release (LOG_INFO,
266 &conf->complaint_cantpublish,
267 "mqtt plugin: successfully reconnected to broker \"%s:%d\"",
268 conf->host, conf->port);
270 return (0);
271 } /* mqtt_reconnect */
273 /* must hold conf->lock when calling. */
274 static int mqtt_connect (mqtt_client_conf_t *conf)
275 {
276 char const *client_id;
277 int status;
279 if (conf->mosq != NULL)
280 return mqtt_reconnect (conf);
282 if (conf->client_id)
283 client_id = conf->client_id;
284 else
285 client_id = hostname_g;
287 #if LIBMOSQUITTO_MAJOR == 0
288 conf->mosq = mosquitto_new (client_id, /* user data = */ conf);
289 #else
290 conf->mosq = mosquitto_new (client_id, conf->clean_session, /* user data = */ conf);
291 #endif
292 if (conf->mosq == NULL)
293 {
294 ERROR ("mqtt plugin: mosquitto_new failed");
295 return (-1);
296 }
298 #if LIBMOSQUITTO_MAJOR != 0
299 if (conf->cacertificatefile) {
300 status = mosquitto_tls_set(conf->mosq, conf->cacertificatefile, NULL,
301 conf->certificatefile, conf->certificatekeyfile, /* pw_callback */NULL);
302 if (status != MOSQ_ERR_SUCCESS) {
303 ERROR ("mqtt plugin: cannot mosquitto_tls_set: %s", mosquitto_strerror(status));
304 mosquitto_destroy (conf->mosq);
305 conf->mosq = NULL;
306 return (-1);
307 }
309 status = mosquitto_tls_opts_set(conf->mosq, SSL_VERIFY_PEER, conf->tlsprotocol, conf->ciphersuite);
310 if (status != MOSQ_ERR_SUCCESS) {
311 ERROR ("mqtt plugin: cannot mosquitto_tls_opts_set: %s", mosquitto_strerror(status));
312 mosquitto_destroy (conf->mosq);
313 conf->mosq = NULL;
314 return (-1);
315 }
317 status = mosquitto_tls_insecure_set(conf->mosq, false);
318 if (status != MOSQ_ERR_SUCCESS) {
319 ERROR ("mqtt plugin: cannot mosquitto_tls_insecure_set: %s", mosquitto_strerror(status));
320 mosquitto_destroy (conf->mosq);
321 conf->mosq = NULL;
322 return (-1);
323 }
324 }
325 #endif
327 if (conf->username && conf->password)
328 {
329 status = mosquitto_username_pw_set (conf->mosq, conf->username, conf->password);
330 if (status != MOSQ_ERR_SUCCESS)
331 {
332 char errbuf[1024];
333 ERROR ("mqtt plugin: mosquitto_username_pw_set failed: %s",
334 (status == MOSQ_ERR_ERRNO)
335 ? sstrerror (errno, errbuf, sizeof (errbuf))
336 : mosquitto_strerror (status));
338 mosquitto_destroy (conf->mosq);
339 conf->mosq = NULL;
340 return (-1);
341 }
342 }
344 #if LIBMOSQUITTO_MAJOR == 0
345 status = mosquitto_connect (conf->mosq, conf->host, conf->port,
346 /* keepalive = */ MQTT_KEEPALIVE, /* clean session = */ conf->clean_session);
347 #else
348 status = mosquitto_connect (conf->mosq, conf->host, conf->port, MQTT_KEEPALIVE);
349 #endif
350 if (status != MOSQ_ERR_SUCCESS)
351 {
352 char errbuf[1024];
353 ERROR ("mqtt plugin: mosquitto_connect failed: %s",
354 (status == MOSQ_ERR_ERRNO)
355 ? sstrerror (errno, errbuf, sizeof (errbuf))
356 : mosquitto_strerror (status));
358 mosquitto_destroy (conf->mosq);
359 conf->mosq = NULL;
360 return (-1);
361 }
363 if (!conf->publish)
364 {
365 mosquitto_message_callback_set (conf->mosq, on_message);
367 status = mosquitto_subscribe (conf->mosq,
368 /* message_id = */ NULL,
369 conf->topic, conf->qos);
370 if (status != MOSQ_ERR_SUCCESS)
371 {
372 ERROR ("mqtt plugin: Subscribing to \"%s\" failed: %s",
373 conf->topic, mosquitto_strerror (status));
375 mosquitto_disconnect (conf->mosq);
376 mosquitto_destroy (conf->mosq);
377 conf->mosq = NULL;
378 return (-1);
379 }
380 }
382 conf->connected = 1;
383 return (0);
384 } /* mqtt_connect */
386 static void *subscribers_thread (void *arg)
387 {
388 mqtt_client_conf_t *conf = arg;
389 int status;
391 conf->loop = 1;
393 while (conf->loop)
394 {
395 status = mqtt_connect (conf);
396 if (status != 0)
397 {
398 sleep (1);
399 continue;
400 }
402 /* The documentation says "0" would map to the default (1000ms), but
403 * that does not work on some versions. */
404 #if LIBMOSQUITTO_MAJOR == 0
405 status = mosquitto_loop (conf->mosq, /* timeout = */ 1000 /* ms */);
406 #else
407 status = mosquitto_loop (conf->mosq,
408 /* timeout[ms] = */ 1000,
409 /* max_packets = */ 100);
410 #endif
411 if (status == MOSQ_ERR_CONN_LOST)
412 {
413 conf->connected = 0;
414 continue;
415 }
416 else if (status != MOSQ_ERR_SUCCESS)
417 {
418 ERROR ("mqtt plugin: mosquitto_loop failed: %s",
419 mosquitto_strerror (status));
420 mosquitto_destroy (conf->mosq);
421 conf->mosq = NULL;
422 conf->connected = 0;
423 continue;
424 }
426 DEBUG ("mqtt plugin: mosquitto_loop succeeded.");
427 } /* while (conf->loop) */
429 pthread_exit (0);
430 } /* void *subscribers_thread */
432 static int publish (mqtt_client_conf_t *conf, char const *topic,
433 void const *payload, size_t payload_len)
434 {
435 int status;
437 pthread_mutex_lock (&conf->lock);
439 status = mqtt_connect (conf);
440 if (status != 0) {
441 pthread_mutex_unlock (&conf->lock);
442 ERROR ("mqtt plugin: unable to reconnect to broker");
443 return (status);
444 }
446 status = mosquitto_publish(conf->mosq, /* message_id */ NULL, topic,
447 #if LIBMOSQUITTO_MAJOR == 0
448 (uint32_t) payload_len, payload,
449 #else
450 (int) payload_len, payload,
451 #endif
452 conf->qos, conf->retain);
453 if (status != MOSQ_ERR_SUCCESS)
454 {
455 char errbuf[1024];
456 c_complain (LOG_ERR,
457 &conf->complaint_cantpublish,
458 "mqtt plugin: mosquitto_publish failed: %s",
459 (status == MOSQ_ERR_ERRNO)
460 ? sstrerror(errno, errbuf, sizeof (errbuf))
461 : mosquitto_strerror(status));
462 /* Mark our connection "down" regardless of the error as a safety
463 * measure; we will try to reconnect the next time we have to publish a
464 * message */
465 conf->connected = 0;
467 pthread_mutex_unlock (&conf->lock);
468 return (-1);
469 }
471 pthread_mutex_unlock (&conf->lock);
472 return (0);
473 } /* int publish */
475 static int format_topic (char *buf, size_t buf_len,
476 data_set_t const *ds, value_list_t const *vl,
477 mqtt_client_conf_t *conf)
478 {
479 char name[MQTT_MAX_TOPIC_SIZE];
480 int status;
482 if ((conf->topic_prefix == NULL) || (conf->topic_prefix[0] == 0))
483 return (FORMAT_VL (buf, buf_len, vl));
485 status = FORMAT_VL (name, sizeof (name), vl);
486 if (status != 0)
487 return (status);
489 status = ssnprintf (buf, buf_len, "%s/%s", conf->topic_prefix, name);
490 if ((status < 0) || (((size_t) status) >= buf_len))
491 return (ENOMEM);
493 return (0);
494 } /* int format_topic */
496 static int mqtt_write (const data_set_t *ds, const value_list_t *vl,
497 user_data_t *user_data)
498 {
499 mqtt_client_conf_t *conf;
500 char topic[MQTT_MAX_TOPIC_SIZE];
501 char payload[MQTT_MAX_MESSAGE_SIZE];
502 int status = 0;
504 if ((user_data == NULL) || (user_data->data == NULL))
505 return (EINVAL);
506 conf = user_data->data;
508 status = format_topic (topic, sizeof (topic), ds, vl, conf);
509 if (status != 0)
510 {
511 ERROR ("mqtt plugin: format_topic failed with status %d.", status);
512 return (status);
513 }
515 status = format_values (payload, sizeof (payload),
516 ds, vl, conf->store_rates);
517 if (status != 0)
518 {
519 ERROR ("mqtt plugin: format_values failed with status %d.", status);
520 return (status);
521 }
523 status = publish (conf, topic, payload, strlen (payload) + 1);
524 if (status != 0)
525 {
526 ERROR ("mqtt plugin: publish failed: %s", mosquitto_strerror (status));
527 return (status);
528 }
530 return (status);
531 } /* mqtt_write */
533 /*
534 * <Publish "name">
535 * Host "example.com"
536 * Port 1883
537 * ClientId "collectd"
538 * User "guest"
539 * Password "secret"
540 * Prefix "collectd"
541 * StoreRates true
542 * Retain false
543 * QoS 0
544 * CACert "ca.pem" Enables TLS if set
545 * CertificateFile "client-cert.pem" optional
546 * CertificateKeyFile "client-key.pem" optional
547 * TLSProtocol "tlsv1.2" optional
548 * </Publish>
549 */
550 static int mqtt_config_publisher (oconfig_item_t *ci)
551 {
552 mqtt_client_conf_t *conf;
553 char cb_name[1024];
554 user_data_t user_data;
555 int status;
556 int i;
558 conf = calloc (1, sizeof (*conf));
559 if (conf == NULL)
560 {
561 ERROR ("mqtt plugin: malloc failed.");
562 return (-1);
563 }
564 conf->publish = 1;
566 conf->name = NULL;
567 status = cf_util_get_string (ci, &conf->name);
568 if (status != 0)
569 {
570 mqtt_free (conf);
571 return (status);
572 }
574 conf->host = strdup (MQTT_DEFAULT_HOST);
575 conf->port = MQTT_DEFAULT_PORT;
576 conf->client_id = NULL;
577 conf->qos = 0;
578 conf->topic_prefix = strdup (MQTT_DEFAULT_TOPIC_PREFIX);
579 conf->store_rates = 1;
581 status = pthread_mutex_init (&conf->lock, NULL);
582 if (status != 0)
583 {
584 mqtt_free (conf);
585 return (status);
586 }
588 C_COMPLAIN_INIT (&conf->complaint_cantpublish);
590 for (i = 0; i < ci->children_num; i++)
591 {
592 oconfig_item_t *child = ci->children + i;
593 if (strcasecmp ("Host", child->key) == 0)
594 cf_util_get_string (child, &conf->host);
595 else if (strcasecmp ("Port", child->key) == 0)
596 {
597 int tmp = cf_util_get_port_number (child);
598 if (tmp < 0)
599 ERROR ("mqtt plugin: Invalid port number.");
600 else
601 conf->port = tmp;
602 }
603 else if (strcasecmp ("ClientId", child->key) == 0)
604 cf_util_get_string (child, &conf->client_id);
605 else if (strcasecmp ("User", child->key) == 0)
606 cf_util_get_string (child, &conf->username);
607 else if (strcasecmp ("Password", child->key) == 0)
608 cf_util_get_string (child, &conf->password);
609 else if (strcasecmp ("QoS", child->key) == 0)
610 {
611 int tmp = -1;
612 status = cf_util_get_int (child, &tmp);
613 if ((status != 0) || (tmp < 0) || (tmp > 2))
614 ERROR ("mqtt plugin: Not a valid QoS setting.");
615 else
616 conf->qos = tmp;
617 }
618 else if (strcasecmp ("Prefix", child->key) == 0)
619 cf_util_get_string (child, &conf->topic_prefix);
620 else if (strcasecmp ("StoreRates", child->key) == 0)
621 cf_util_get_boolean (child, &conf->store_rates);
622 else if (strcasecmp ("Retain", child->key) == 0)
623 cf_util_get_boolean (child, &conf->retain);
624 else if (strcasecmp ("CACert", child->key) == 0)
625 cf_util_get_string (child, &conf->cacertificatefile);
626 else if (strcasecmp ("CertificateFile", child->key) == 0)
627 cf_util_get_string (child, &conf->certificatefile);
628 else if (strcasecmp ("CertificateKeyFile", child->key) == 0)
629 cf_util_get_string (child, &conf->certificatekeyfile);
630 else if (strcasecmp ("TLSProtocol", child->key) == 0)
631 cf_util_get_string (child, &conf->tlsprotocol);
632 else if (strcasecmp ("CipherSuite", child->key) == 0)
633 cf_util_get_string (child, &conf->ciphersuite);
634 else
635 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
636 }
638 ssnprintf (cb_name, sizeof (cb_name), "mqtt/%s", conf->name);
639 memset (&user_data, 0, sizeof (user_data));
640 user_data.data = conf;
642 plugin_register_write (cb_name, mqtt_write, &user_data);
643 return (0);
644 } /* mqtt_config_publisher */
646 /*
647 * <Subscribe "name">
648 * Host "example.com"
649 * Port 1883
650 * ClientId "collectd"
651 * User "guest"
652 * Password "secret"
653 * Topic "collectd/#"
654 * </Subscribe>
655 */
656 static int mqtt_config_subscriber (oconfig_item_t *ci)
657 {
658 mqtt_client_conf_t **tmp;
659 mqtt_client_conf_t *conf;
660 int status;
661 int i;
663 conf = calloc (1, sizeof (*conf));
664 if (conf == NULL)
665 {
666 ERROR ("mqtt plugin: malloc failed.");
667 return (-1);
668 }
669 conf->publish = 0;
671 conf->name = NULL;
672 status = cf_util_get_string (ci, &conf->name);
673 if (status != 0)
674 {
675 mqtt_free (conf);
676 return (status);
677 }
679 conf->host = strdup (MQTT_DEFAULT_HOST);
680 conf->port = MQTT_DEFAULT_PORT;
681 conf->client_id = NULL;
682 conf->qos = 2;
683 conf->topic = strdup (MQTT_DEFAULT_TOPIC);
684 conf->clean_session = 1;
686 status = pthread_mutex_init (&conf->lock, NULL);
687 if (status != 0)
688 {
689 mqtt_free (conf);
690 return (status);
691 }
693 C_COMPLAIN_INIT (&conf->complaint_cantpublish);
695 for (i = 0; i < ci->children_num; i++)
696 {
697 oconfig_item_t *child = ci->children + i;
698 if (strcasecmp ("Host", child->key) == 0)
699 cf_util_get_string (child, &conf->host);
700 else if (strcasecmp ("Port", child->key) == 0)
701 {
702 int tmp = cf_util_get_port_number (child);
703 if (tmp < 0)
704 ERROR ("mqtt plugin: Invalid port number.");
705 else
706 conf->port = tmp;
707 }
708 else if (strcasecmp ("ClientId", child->key) == 0)
709 cf_util_get_string (child, &conf->client_id);
710 else if (strcasecmp ("User", child->key) == 0)
711 cf_util_get_string (child, &conf->username);
712 else if (strcasecmp ("Password", child->key) == 0)
713 cf_util_get_string (child, &conf->password);
714 else if (strcasecmp ("QoS", child->key) == 0)
715 {
716 int tmp = -1;
717 status = cf_util_get_int (child, &tmp);
718 if ((status != 0) || (tmp < 0) || (tmp > 2))
719 ERROR ("mqtt plugin: Not a valid QoS setting.");
720 else
721 conf->qos = tmp;
722 }
723 else if (strcasecmp ("Topic", child->key) == 0)
724 cf_util_get_string (child, &conf->topic);
725 else if (strcasecmp ("CleanSession", child->key) == 0)
726 cf_util_get_boolean (child, &conf->clean_session);
727 else
728 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
729 }
731 tmp = realloc (subscribers, sizeof (*subscribers) * subscribers_num);
732 if (tmp == NULL)
733 {
734 ERROR ("mqtt plugin: realloc failed.");
735 mqtt_free (conf);
736 return (-1);
737 }
738 subscribers = tmp;
739 subscribers[subscribers_num] = conf;
740 subscribers_num++;
742 return (0);
743 } /* mqtt_config_subscriber */
745 /*
746 * <Plugin mqtt>
747 * <Publish "name">
748 * # ...
749 * </Publish>
750 * <Subscribe "name">
751 * # ...
752 * </Subscribe>
753 * </Plugin>
754 */
755 static int mqtt_config (oconfig_item_t *ci)
756 {
757 int i;
759 for (i = 0; i < ci->children_num; i++)
760 {
761 oconfig_item_t *child = ci->children + i;
763 if (strcasecmp ("Publish", child->key) == 0)
764 mqtt_config_publisher (child);
765 else if (strcasecmp ("Subscribe", child->key) == 0)
766 mqtt_config_subscriber (child);
767 else
768 ERROR ("mqtt plugin: Unknown config option: %s", child->key);
769 }
771 return (0);
772 } /* int mqtt_config */
774 static int mqtt_init (void)
775 {
776 size_t i;
778 mosquitto_lib_init ();
780 for (i = 0; i < subscribers_num; i++)
781 {
782 int status;
784 if (subscribers[i]->loop)
785 continue;
787 status = plugin_thread_create (&subscribers[i]->thread,
788 /* attrs = */ NULL,
789 /* func = */ subscribers_thread,
790 /* args = */ subscribers[i]);
791 if (status != 0)
792 {
793 char errbuf[1024];
794 ERROR ("mqtt plugin: pthread_create failed: %s",
795 sstrerror (errno, errbuf, sizeof (errbuf)));
796 continue;
797 }
798 }
800 return (0);
801 } /* mqtt_init */
803 void module_register (void)
804 {
805 plugin_register_complex_config ("mqtt", mqtt_config);
806 plugin_register_init ("mqtt", mqtt_init);
807 } /* void module_register */
809 /* vim: set sw=4 sts=4 et fdm=marker : */