Repository: wso2/micro-integrator
Issue: RabbitMq Inbound Ep message consumption is not paused when the underlying task is stopped.

**Description:**
When a node looses connection to the cluster, the rabbit mq inbound task is paused, however the inbound endpoint continues to receive the message and not paused. This need to be fixed.

Changed files:
components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/InboundOneTimeTriggerRequestProcessor.java
components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/rabbitmq/RabbitMQTask.java
components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/OneTimeTriggerInboundTask.java

diff --git a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/InboundOneTimeTriggerRequestProcessor.java b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/InboundOneTimeTriggerRequestProcessor.java
index f1625fb80..a630468b9 100644
--- a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/InboundOneTimeTriggerRequestProcessor.java
+++ b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/InboundOneTimeTriggerRequestProcessor.java
@@ -24,8 +24,10 @@ import org.apache.synapse.core.SynapseEnvironment;
 import org.apache.synapse.inbound.InboundRequestProcessor;
 import org.apache.synapse.startup.quartz.StartUpController;
 import org.apache.synapse.task.TaskDescription;
+import org.apache.synapse.task.TaskManager;
 import org.wso2.carbon.inbound.endpoint.persistence.InboundEndpointsDataStore;
-
+import org.wso2.carbon.inbound.endpoint.protocol.rabbitmq.RabbitMQTask;
+import org.wso2.micro.integrator.mediation.ntask.NTaskTaskManager;
 
 import static org.wso2.carbon.inbound.endpoint.common.Constants.SUPER_TENANT_DOMAIN_NAME;
 
@@ -76,6 +78,14 @@ public abstract class InboundOneTimeTriggerRequestProcessor implements InboundRe
                 startUpController = new StartUpController();
                 startUpController.setTaskDescription(taskDescription);
                 startUpController.init(synapseEnvironment);
+                // registering a listener to identify task removal or deletions.
+                if (task instanceof RabbitMQTask) {
+                    TaskManager taskManagerImpl = synapseEnvironment.getTaskManager().getTaskManagerImpl();
+                    if (taskManagerImpl instanceof NTaskTaskManager) {
+                        ((NTaskTaskManager) taskManagerImpl).registerListener((RabbitMQTask) task,
+                                                                              taskDescription.getName());
+                    }
+                }
             } catch (Exception e) {
                 log.error("Error starting the inbound endpoint " + name + ". Unable to schedule the task. " + e
                         .getLocalizedMessage(), e);
diff --git a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/OneTimeTriggerInboundTask.java b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/OneTimeTriggerInboundTask.java
index 2893e36f4..5f4555043 100644
--- a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/OneTimeTriggerInboundTask.java
+++ b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/common/OneTimeTriggerInboundTask.java
@@ -30,6 +30,8 @@ public abstract class OneTimeTriggerInboundTask implements org.apache.synapse.ta
     private static final Log logger = LogFactory.getLog(InboundTask.class.getName());
     private boolean isOneTimeTriggered = false;
     private OneTimeTriggerAbstractCallback callback;
+    // boolean used to identify the re-trigger of the task.
+    private boolean reTrigger = false;
 
     public void execute() {
         //this check is there to synchronize task cycle round hit and connection lost reconnection
@@ -38,15 +40,11 @@ public abstract class OneTimeTriggerInboundTask implements org.apache.synapse.ta
         if (callback != null && callback.isCallbackSuspended()) {
             callback.releaseCallbackSuspension();
         }
-
-        if (!isOneTimeTriggered) {
+        if (!isOneTimeTriggered || reTrigger) {
             isOneTimeTriggered = true;
-            if (logger.isDebugEnabled()) {
-                logger.debug("Common One time trigger Inbound Task executing.");
-            }
+            reTrigger = false;
+            logger.debug("Common One time trigger Inbound Task executing.");
             taskExecute();
-        } else {
-            return;
         }
     }
 
@@ -58,5 +56,10 @@ public abstract class OneTimeTriggerInboundTask implements org.apache.synapse.ta
         return callback;
     }
 
+    protected void setReTrigger() {
+        logger.debug("Enabling re-trigger.");
+        this.reTrigger = true;
+    }
+
     protected abstract void taskExecute();
 }
diff --git a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/rabbitmq/RabbitMQTask.java b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/rabbitmq/RabbitMQTask.java
index 9300153c1..65d0044a5 100644
--- a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/rabbitmq/RabbitMQTask.java
+++ b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/rabbitmq/RabbitMQTask.java
@@ -22,8 +22,9 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.synapse.core.SynapseEnvironment;
 import org.wso2.carbon.inbound.endpoint.common.OneTimeTriggerInboundTask;
+import org.wso2.micro.integrator.ntask.core.impl.LocalTaskActionListener;
 
-public class RabbitMQTask extends OneTimeTriggerInboundTask {
+public class RabbitMQTask extends OneTimeTriggerInboundTask implements LocalTaskActionListener {
 
     private static final Log log = LogFactory.getLog(RabbitMQTask.class.getName());
 
@@ -46,4 +47,10 @@ public class RabbitMQTask extends OneTimeTriggerInboundTask {
     public void destroy() {
         log.debug("Destroying.");
     }
+
+    @Override
+    public void notifyLocalTaskRemoval(String taskName) {
+        setReTrigger();
+        rabbitMQConnectionConsumer.requestShutdown();
+    }
 }