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