few changes added to transport adapters after testing with the cluster

components/iot-plugins/iot-base-plugin/org.wso2.carbon.device.mgt.iot.input.adapter.mqtt/src/main/java/org/wso2/carbon/device/mgt/iot/input/adapter/mqtt/MQTTEventAdapter.java

@@ -96,9 +96,12 @@ public class MQTTEventAdapter implements InputEventAdapter {
 
     @Override
     public void connect() {
+        if (!mqttAdapterListener.isConnectionInitialized()) {
             mqttAdapterListener.createConnection();
         }
 
+    }
+
     @Override
     public void disconnect() {
         //when mqtt and this feature  both together then this method becomes a blocking method, Therefore
@@ -147,7 +150,7 @@ public class MQTTEventAdapter implements InputEventAdapter {
 
     @Override
     public boolean isPolling() {
-        return true;
+        return mqttAdapterListener.isConnectionInitialized();
     }
 
 }

components/iot-plugins/iot-base-plugin/org.wso2.carbon.device.mgt.iot.input.adapter.mqtt/src/main/java/org/wso2/carbon/device/mgt/iot/input/adapter/mqtt/util/MQTTAdapterListener.java

@@ -58,6 +58,7 @@ public class MQTTAdapterListener implements MqttCallback, Runnable {
     private MqttClient mqttClient;
     private MqttConnectOptions connectionOptions;
     private boolean cleanSession;
+    private boolean connectionInitialized;
 
     private MQTTBrokerConnectionConfiguration mqttBrokerConnectionConfiguration;
     private String topic;
@@ -279,6 +280,11 @@ public class MQTTAdapterListener implements MqttCallback, Runnable {
     }
 
     public void createConnection() {
+        connectionInitialized = true;
         new Thread(this).start();
     }
+
+    public boolean isConnectionInitialized() {
+        return connectionInitialized;
+    }
 }

components/iot-plugins/iot-base-plugin/org.wso2.carbon.device.mgt.iot.input.adapter.xmpp/src/main/java/org/wso2/carbon/device/mgt/iot/input/adapter/xmpp/XMPPEventAdapter.java

@@ -103,8 +103,10 @@ public class XMPPEventAdapter implements InputEventAdapter {
 
     @Override
     public void connect() {
+        if (xmppAdapterListener.isConnectionInitialized()) {
             xmppAdapterListener.createConnection();
         }
+    }
 
     @Override
     public void disconnect() {
@@ -142,7 +144,7 @@ public class XMPPEventAdapter implements InputEventAdapter {
 
     @Override
     public boolean isPolling() {
-        return true;
+        return xmppAdapterListener.isConnectionInitialized();
     }
 
 }

components/iot-plugins/iot-base-plugin/org.wso2.carbon.device.mgt.iot.input.adapter.xmpp/src/main/java/org/wso2/carbon/device/mgt/iot/input/adapter/xmpp/util/XMPPAdapterListener.java

@@ -21,6 +21,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.jivesoftware.smack.ConnectionConfiguration;
 import org.jivesoftware.smack.PacketListener;
+import org.jivesoftware.smack.ReconnectionManager;
 import org.jivesoftware.smack.SmackConfiguration;
 import org.jivesoftware.smack.XMPPConnection;
 import org.jivesoftware.smack.XMPPException;
@@ -53,6 +54,7 @@ public class XMPPAdapterListener implements Runnable {
     private ContentValidator contentValidator;
     private ContentTransformer contentTransformer;
     private PacketListener packetListener;
+    private boolean connectionInitialized;
 
     private InputEventAdapterListener eventAdapterListener = null;
 
@@ -217,6 +219,11 @@ public class XMPPAdapterListener implements Runnable {
     }
 
     public void createConnection() {
+        connectionInitialized = true;
         new Thread(this).start();
     }
+
+    public boolean isConnectionInitialized() {
+        return connectionInitialized;
+    }
 }

components/iot-plugins/virtual-fire-alarm-plugin/org.wso2.carbon.device.mgt.iot.virtualfirealarm.plugin/src/main/java/org/wso2/carbon/device/mgt/iot/virtualfirealarm/plugin/impl/util/VirtualFirealarmStartupListener.java

@@ -21,6 +21,7 @@ package org.wso2.carbon.device.mgt.iot.virtualfirealarm.plugin.impl.util;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.wso2.carbon.core.ServerStartupObserver;
+import org.wso2.carbon.device.mgt.iot.virtualfirealarm.plugin.internal.VirtualFirealarmManagementDataHolder;
 
 import java.io.IOException;
 
@@ -36,7 +37,7 @@ public class VirtualFirealarmStartupListener implements ServerStartupObserver {
         try {
             VirtualFireAlarmUtils.setupMqttInputAdapter();
             VirtualFireAlarmUtils.setupXmppInputAdapter();
-
+            VirtualFirealarmManagementDataHolder.getInstance().getInputEventAdapterService().start();
         } catch (IOException e) {
             log.error("Failed to intilaize the virtual firealarm input adapter", e);
         }