From 85bc1838491f46afd8ef1aaa610a114775fa0829 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 11 Jun 2018 14:29:06 -0400
Subject: [PATCH] INT-4486: Properly implement stop(Runnable)

JIRA: https://jira.spring.io/browse/INT-4486

The `SmartLifecycle.stop(Runnable callback)` must always call the
`callback` in the end independently of the internal state

* Revise all the `SmartLifecycle` implementations for the proper
`callback` handling

**Cherry-pick to 5.0.x and 4.3.x**
---
 .../AbstractSubscribableAmqpChannel.java      |  5 +++-
 .../amqp/config/AmqpChannelFactoryBean.java   | 14 +++++-----
 .../config/ConsumerEndpointFactoryBean.java   |  3 +++
 ...ourcePollingChannelAdapterFactoryBean.java |  5 +++-
 .../dsl/IntegrationFlowAdapter.java           |  3 +++
 .../endpoint/AbstractEndpoint.java            |  5 +++-
 ...eTailInboundChannelAdapterFactoryBean.java |  6 ++++-
 .../jms/SubscribableJmsChannel.java           | 21 ++++++++++-----
 .../jms/config/JmsChannelFactoryBean.java     |  5 +++-
 .../channel/SubscribableRedisChannel.java     | 27 +++++++------------
 ...logReceivingChannelAdapterFactoryBean.java |  8 +++++-
 11 files changed, 67 insertions(+), 35 deletions(-)

diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
index 44191c99ed..b6a406fe5d 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -234,6 +234,9 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 			this.container.stop(callback);
 			this.declared = false;
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
index 5b86987dbc..426e542ff9 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -66,10 +66,11 @@ import org.springframework.util.StringUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.1
  */
-public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel> implements SmartLifecycle,
-		DisposableBean, BeanNameAware {
+public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel>
+		implements SmartLifecycle, DisposableBean, BeanNameAware {
 
 	private volatile AbstractAmqpChannel channel;
 
@@ -516,13 +517,14 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		if (this.channel instanceof SmartLifecycle) {
 			((SmartLifecycle) this.channel).stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
 	protected void destroyInstance(AbstractAmqpChannel instance) throws Exception {
-		if (instance instanceof DisposableBean) {
-			((DisposableBean) this.channel).destroy();
-		}
+		this.channel.destroy();
 	}
 
 }
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
index 7203aa9828..0ed43846a0 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
@@ -375,6 +375,9 @@ public class ConsumerEndpointFactoryBean
 		if (this.endpoint != null) {
 			this.endpoint.stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
index 8d3a3aef77..a19abfaefa 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -247,6 +247,9 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 		if (this.adapter != null) {
 			this.adapter.stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java
index 74a6abad6d..615e7ca5b2 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java
@@ -98,6 +98,9 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
 		if (this.running.getAndSet(false)) {
 			this.targetIntegrationFlow.stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
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 64041fcf56..a98cfa6fe3 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -199,6 +199,9 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport
 					logger.info("stopped " + this);
 				}
 			}
+			else {
+				callback.run();
+			}
 		}
 		finally {
 			this.lifecycleLock.unlock();
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java
index 106bbb5325..ac8f56906b 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -36,6 +36,7 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @author Artem Bilan
  * @author Ali Shahbour
+ *
  * @since 3.0
  *
  */
@@ -192,6 +193,9 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
 		if (this.adapter != null) {
 			this.adapter.stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
index 22a9fb6ffc..cc25beca19 100644
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -43,9 +43,12 @@ import org.springframework.util.Assert;
 /**
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
-public class SubscribableJmsChannel extends AbstractJmsChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
+public class SubscribableJmsChannel extends AbstractJmsChannel
+		implements SubscribableChannel, SmartLifecycle, DisposableBean {
 
 	private final AbstractMessageListenerContainer container;
 
@@ -72,13 +75,15 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 
 	@Override
 	public boolean subscribe(MessageHandler handler) {
-		Assert.state(this.dispatcher != null, "'MessageDispatcher' must not be null. This channel might not have been initialized");
+		Assert.state(this.dispatcher != null,
+				"'MessageDispatcher' must not be null. This channel might not have been initialized");
 		return this.dispatcher.addHandler(handler);
 	}
 
 	@Override
 	public boolean unsubscribe(MessageHandler handler) {
-		Assert.state(this.dispatcher != null, "'MessageDispatcher' must not be null. This channel might not have been initialized");
+		Assert.state(this.dispatcher != null,
+				"'MessageDispatcher' must not be null. This channel might not have been initialized");
 		return this.dispatcher.removeHandler(handler);
 	}
 
@@ -126,7 +131,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 
 	@Override
 	public boolean isAutoStartup() {
-		return (this.container != null) ? this.container.isAutoStartup() : false;
+		return (this.container != null) && this.container.isAutoStartup();
 	}
 
 	@Override
@@ -136,7 +141,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 
 	@Override
 	public boolean isRunning() {
-		return (this.container != null) ? this.container.isRunning() : false;
+		return (this.container != null) && this.container.isRunning();
 	}
 
 	@Override
@@ -158,6 +163,9 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 		if (this.container != null) {
 			this.container.stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
@@ -226,6 +234,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
 				throw new MessagingException("failed to handle incoming JMS Message", e);
 			}
 		}
+
 	}
 
 }
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java
index ee0365239f..a80757c7a8 100644
--- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java
+++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -548,6 +548,9 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
 		if (this.channel instanceof SubscribableJmsChannel) {
 			((SubscribableJmsChannel) this.channel).stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override
diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
index 887fab1642..9bb149294c 100644
--- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
+++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -51,6 +51,7 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 @SuppressWarnings("rawtypes")
@@ -105,7 +106,6 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
 	/**
 	 * Specify the maximum number of subscribers supported by the
 	 * channel's dispatcher.
-	 *
 	 * @param maxSubscribers The maximum number of subscribers allowed.
 	 */
 	public void setMaxSubscribers(int maxSubscribers) {
@@ -168,45 +168,37 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
 
 	@Override
 	public boolean isAutoStartup() {
-		return (this.container != null) && this.container.isAutoStartup();
+		return this.container.isAutoStartup();
 	}
 
 	@Override
 	public int getPhase() {
-		return (this.container != null) ? this.container.getPhase() : 0;
+		return this.container.getPhase();
 	}
 
 	@Override
 	public boolean isRunning() {
-		return (this.container != null) && this.container.isRunning();
+		return this.container.isRunning();
 	}
 
 	@Override
 	public void start() {
-		if (this.container != null) {
-			this.container.start();
-		}
+		this.container.start();
 	}
 
 	@Override
 	public void stop() {
-		if (this.container != null) {
-			this.container.stop();
-		}
+		this.container.stop();
 	}
 
 	@Override
 	public void stop(Runnable callback) {
-		if (this.container != null) {
-			this.container.stop(callback);
-		}
+		this.container.stop(callback);
 	}
 
 	@Override
 	public void destroy() throws Exception {
-		if (this.container != null) {
-			this.container.destroy();
-		}
+		this.container.destroy();
 	}
 
 	private class MessageListenerDelegate {
@@ -230,6 +222,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
 						+ "' (" + SubscribableRedisChannel.this.getFullChannelName() + ").", e);
 			}
 		}
+
 	}
 
 }
diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java
index b011b1b5cd..e019603413 100644
--- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java
+++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,7 +32,10 @@ import org.springframework.util.Assert;
 
 /**
  * Factory bean to create syslog inbound adapters (UDP or TCP).
+ *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 3.0
  *
  */
@@ -163,6 +166,9 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea
 		if (this.adapter != null) {
 			this.adapter.stop(callback);
 		}
+		else {
+			callback.run();
+		}
 	}
 
 	@Override