add MqttReconnectCallback

Author: Szymon Lisiecki

src/main/java/com/orange/lo/sdk/mqtt/AbstractDataManagementMqtt.java

@@ -1,23 +1,41 @@
+/**
+ * Copyright (c) Orange. All Rights Reserved.
+ * <p>
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
 package com.orange.lo.sdk.mqtt;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.orange.lo.sdk.LOApiClientParameters;
-import com.orange.lo.sdk.mqtt.exceptions.LoMqttException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
 import org.eclipse.paho.client.mqttv3.IMqttClient;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
+import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.orange.lo.sdk.LOApiClientParameters;
+import com.orange.lo.sdk.mqtt.exceptions.LoMqttException;
+
 
 public abstract class AbstractDataManagementMqtt {
 
     private final IMqttClient mqttClient;
     private final ObjectMapper objectMapper;
     private final LOApiClientParameters parameters;
+	private final MqttReconnectCallback mqttReconnectCallback;
 
     public AbstractDataManagementMqtt(LOApiClientParameters parameters, MqttClientFactory mqttClientFactory) {
+    	this.mqttReconnectCallback = new MqttReconnectCallback();
         this.mqttClient = mqttClientFactory.getMqttClient();
+        this.mqttClient.setCallback(mqttReconnectCallback);
         this.parameters = parameters;
         this.objectMapper = new ObjectMapper();
     }
@@ -56,16 +74,13 @@ protected MqttConnectOptions getMqttConnectionOptions() {
     }
 
     protected void subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) {
-        try {
-            mqttClient.subscribe(topicFilter, qos, messageListener);
-        } catch (MqttException e) {
-            throw new LoMqttException(e);
-        }
+        subscribe(new String[] {topicFilter}, new int[] {qos}, new IMqttMessageListener[] {messageListener});
     }
 
     protected void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) {
         try {
             mqttClient.subscribe(topicFilters, qos, messageListeners);
+            mqttReconnectCallback.addSubscriptions(topicFilters, qos, messageListeners);
         } catch (MqttException e) {
             throw new LoMqttException(e);
         }
@@ -86,4 +101,35 @@ protected ObjectMapper getObjectMapper() {
     protected LOApiClientParameters getParameters() {
         return parameters;
     }
+    
+    class MqttReconnectCallback implements MqttCallbackExtended {
+
+		private final List<String>  topicFiltersList = new ArrayList<>();
+		private final List<Integer> qosList = new ArrayList<>();
+		private final List<IMqttMessageListener>  listenersList = new ArrayList<>();
+
+		public void addSubscriptions(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) {
+			topicFiltersList.addAll(Arrays.asList(topicFilters));
+			qosList.addAll(Arrays.stream(qos).boxed().collect(Collectors.toList()));
+			listenersList.addAll(Arrays.asList(messageListeners));
+		}
+
+		public void connectionLost(Throwable cause) {
+		}
+
+		public void messageArrived(String topic, MqttMessage message) throws Exception {
+		}
+
+		public void deliveryComplete(IMqttDeliveryToken token) {
+		}
+
+		public void connectComplete(boolean reconnect, String serverURI) {
+			if(reconnect) {
+				subscribe(
+						topicFiltersList.toArray(new String[0]), 
+						qosList.stream().mapToInt(Integer::intValue).toArray(),
+						listenersList.toArray(new IMqttMessageListener[0]));
+			}
+		}
+	}
 }