MQTT: Resubscribe after a reconnect

After losing our connection to a MQTT broker our subscription will be gone,
too, so we need to subscribe to our topic once again after reconnecting,
otherwise collectd will not receive any MQTT messages anymore.

Signed-off-by: Maciej S. Szmigiero <mail@maciej.szmigiero.name>

diff --git a/src/mqtt.c b/src/mqtt.c
index 55bb5d93a9954601c6d8dcb9ea18a73219951981..adba2b1528f1518bf943570ad1b3a7d6937d3a89 100644 (file)
--- a/src/mqtt.c
+++ b/src/mqtt.c
@@ -238,6 +238,19 @@ static void on_message(
   sfree(vl.values);
 } /* void on_message */
 
+static int mqtt_subscribe(mqtt_client_conf_t *conf) {
+  int status = mosquitto_subscribe(conf->mosq, /* message_id = */ NULL,
+                                   conf->topic, conf->qos);
+  if (status != MOSQ_ERR_SUCCESS) {
+    ERROR("mqtt plugin: Subscribing to \"%s\" failed: %s", conf->topic,
+          mosquitto_strerror(status));
+    mosquitto_disconnect(conf->mosq);
+    return -1;
+  }
+
+  return 0;
+}
+
 /* must hold conf->lock when calling. */
 static int mqtt_reconnect(mqtt_client_conf_t *conf) {
   int status;
@@ -252,6 +265,12 @@ static int mqtt_reconnect(mqtt_client_conf_t *conf) {
     return -1;
   }
 
+  if (!conf->publish) {
+    status = mqtt_subscribe(conf);
+    if (status != 0)
+      return status;
+  }
+
   conf->connected = true;
 
   c_release(LOG_INFO, &conf->complaint_cantpublish,
@@ -352,17 +371,11 @@ static int mqtt_connect(mqtt_client_conf_t *conf) {
   if (!conf->publish) {
     mosquitto_message_callback_set(conf->mosq, on_message);
 
-    status =
-        mosquitto_subscribe(conf->mosq,
-                            /* message_id = */ NULL, conf->topic, conf->qos);
-    if (status != MOSQ_ERR_SUCCESS) {
-      ERROR("mqtt plugin: Subscribing to \"%s\" failed: %s", conf->topic,
-            mosquitto_strerror(status));
-
-      mosquitto_disconnect(conf->mosq);
+    status = mqtt_subscribe(conf);
+    if (status != 0) {
       mosquitto_destroy(conf->mosq);
       conf->mosq = NULL;
-      return -1;
+      return status;
     }
   }