Merge pull request #325 from mathieu1fb/feature/mqtt-auto-reconnect

java: add MQTT auto reconnect to IoT3 Core

Author: mathieu1fb

java/iot3/core/src/main/java/com/orange/iot3core/IoT3Core.java

@@ -205,12 +205,19 @@ public void mqttPublish(String topic, String message, boolean retain) {
         if(mqttClient != null) mqttClient.publishMessage(topic, message, retain);
     }
 
+    /**
+     * Check that the MQTT connection is established
+     */
+    public boolean isMqttConnected() {
+        if(mqttClient != null) return mqttClient.isConnected();
+        else return false;
+    }
+
     /**
      * Check that the MQTT connection is secured with TLS
      */
     public boolean isMqttConnectionSecured() {
-        if(mqttClient != null) return mqttClient.isConnected() && mqttClient.isConnectionSecured();
-        else return false;
+        return isMqttConnected() && mqttClient.isConnectionSecured();
     }
 
     /**

java/iot3/core/src/main/java/com/orange/iot3core/clients/MqttClient.java

@@ -9,6 +9,8 @@
 
 import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
 import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.lifecycle.MqttClientAutoReconnect;
+import com.hivemq.client.mqtt.lifecycle.MqttClientAutoReconnectBuilder;
 import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
 import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientBuilder;
 import com.hivemq.client.mqtt.mqtt5.datatypes.Mqtt5UserProperties;
@@ -26,6 +28,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -57,6 +60,10 @@ public MqttClient(String serverHost,
                 .identifier(clientId)
                 .serverHost(serverHost)
                 .serverPort(serverPort)
+                .automaticReconnect()
+                        .initialDelay(500, TimeUnit.MILLISECONDS)
+                        .maxDelay(5, TimeUnit.SECONDS)
+                        .applyAutomaticReconnect()
                 .addDisconnectedListener(context1 -> {
                     LOGGER.log(Level.INFO, "Disconnected from MQTT broker " + serverHost);
                     callback.connectionLost(context1.getCause());

java/iot3/mobility/src/main/java/com/orange/iot3mobility/IoT3Mobility.java

@@ -146,6 +146,14 @@ public void mqttUnsubscriptionComplete(Throwable unsubscribeFailure) {
         TrueTime.initTrueTime();
     }
 
+    /**
+     * Check that the connection is established.
+     */
+    public boolean isConnected() {
+        if(ioT3Core != null) return ioT3Core.isMqttConnected();
+        else return false;
+    }
+
     /**
      * Retrieve the IoT3Core instance powering IoT3Mobility.
      */
@@ -324,8 +332,8 @@ public void sendCam(CAM cam) {
         String geoExtension = QuadTileHelper.quadKeyToQuadTopic(quadkey);
         String topic = context + "/inQueue/v2x/cam/" + uuid + geoExtension;
 
-        // send the message
-        if(ioT3Core != null) ioT3Core.mqttPublish(topic, cam.getJsonCAM().toString());
+        // send the message only if the client is connected
+        if(isConnected()) ioT3Core.mqttPublish(topic, cam.getJsonCAM().toString());
     }
 
     /**
@@ -392,7 +400,7 @@ public void sendDenm(DENM denm) {
         String geoExtension = QuadTileHelper.quadKeyToQuadTopic(quadkey);
         String topic = context + "/inQueue/v2x/denm/" + uuid + geoExtension;
 
-        // send the message
+        // send the message even if the client is disconnected, so it will be queued
         if(ioT3Core != null) ioT3Core.mqttPublish(topic, denm.getJsonDENM().toString());
     }
 
@@ -410,8 +418,8 @@ public void sendCpm(CPM cpm) {
         String geoExtension = QuadTileHelper.quadKeyToQuadTopic(quadkey);
         String topic = context + "/inQueue/v2x/cpm/" + uuid + geoExtension;
 
-        // send the message
-        if(ioT3Core != null) ioT3Core.mqttPublish(topic, cpm.getJson().toString());
+        // send the message only if the client is connected
+        if(isConnected()) ioT3Core.mqttPublish(topic, cpm.getJson().toString());
     }
 
     /**