From 7dcb6ce58f63db41505356aa53008755b43ac5a2 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Tue, 4 Nov 2014 10:01:43 -0500
Subject: [PATCH] INT-3457 Fix stop(Runnable) in AbstractEndpoint

JIRA: https://jira.spring.io/browse/INT-3547
JIRA: https://jira.spring.io/browse/INT-3546

INT-3486 changed stop(Runnable) so that it could be
overridden by subclasses, to allow separation of the
`stop()` and `callback` invocation.

However, the refactoring changed the logic such that
the `running` field is not reset, causing `doStop()` to
be called even when the component was not running.

Reset the running field.

Also, the MQTT inbound channel adapter did not test
for a `null` `client` in `doStop()`.
---
 .../endpoint/AbstractEndpoint.java            | 20 +++++++--
 .../MqttPahoMessageDrivenChannelAdapter.java  | 44 ++++++++++---------
 2 files changed, 40 insertions(+), 24 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
index 4594cc6836..7c3e0fdbad 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
@@ -35,6 +35,7 @@ import org.springframework.scheduling.TaskScheduler;
  *
  * @author Mark Fisher
  * @author Kris Jacyna
+ * @author Gary Russell
  */
 public abstract class AbstractEndpoint extends IntegrationObjectSupport implements SmartLifecycle {
 
@@ -57,20 +58,24 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
 		this.phase = phase;
 	}
 
+	@Override
 	public void setTaskScheduler(TaskScheduler taskScheduler) {
 		super.setTaskScheduler(taskScheduler);
 	}
 
 	// SmartLifecycle implementation
 
+	@Override
 	public final boolean isAutoStartup() {
 		return this.autoStartup;
 	}
 
+	@Override
 	public final int getPhase() {
 		return this.phase;
 	}
 
+	@Override
 	public final boolean isRunning() {
 		this.lifecycleLock.lock();
 		try {
@@ -81,6 +86,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
 		}
 	}
 
+	@Override
 	public final void start() {
 		this.lifecycleLock.lock();
 		try {
@@ -97,6 +103,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
 		}
 	}
 
+	@Override
 	public final void stop() {
 		this.lifecycleLock.lock();
 		try {
@@ -113,10 +120,17 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
 		}
 	}
 
+	@Override
 	public final void stop(Runnable callback) {
 		this.lifecycleLock.lock();
 		try {
-			doStop(callback);
+			if (this.running) {
+				doStop(callback);
+				this.running = false;
+				if (logger.isInfoEnabled()) {
+					logger.info("stopped " + this);
+				}
+			}
 		}
 		finally {
 			this.lifecycleLock.unlock();
@@ -128,8 +142,8 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport implemen
 	 * @param callback the Runnable to invoke.
 	 */
 	protected void doStop(Runnable callback) {
-	    doStop();
-	    callback.run();
+		doStop();
+		callback.run();
 	}
 
 	/**
diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
index 9c1db20a6c..b802bf6aef 100644
--- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
+++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
@@ -119,28 +119,30 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 	protected void doStop() {
 		this.cancelReconnect();
 		super.doStop();
-		try {
-			this.client.unsubscribe(this.getTopic())
-					.waitForCompletion(this.completionTimeout);
+		if (this.client != null) {
+			try {
+				this.client.unsubscribe(this.getTopic())
+						.waitForCompletion(this.completionTimeout);
+			}
+			catch (MqttException e) {
+				logger.error("Exception while unsubscribing", e);
+			}
+			try {
+				this.client.disconnect()
+						.waitForCompletion(this.completionTimeout);
+			}
+			catch (MqttException e) {
+				logger.error("Exception while disconnecting", e);
+			}
+			try {
+				this.client.close();
+			}
+			catch (MqttException e) {
+				logger.error("Exception while closing", e);
+			}
+			this.connected = false;
+			this.client = null;
 		}
-		catch (MqttException e) {
-			logger.error("Exception while unsubscribing", e);
-		}
-		try {
-			this.client.disconnect()
-					.waitForCompletion(this.completionTimeout);
-		}
-		catch (MqttException e) {
-			logger.error("Exception while disconnecting", e);
-		}
-		try {
-			this.client.close();
-		}
-		catch (MqttException e) {
-			logger.error("Exception while closing", e);
-		}
-		this.connected = false;
-		this.client = null;
 	}
 
 	@Override