INT-4307: JMS: Endpoint Stop: Shutdown Container

garyrussell · artembilan · commit 76b1d1df06ea · 2018-09-13T14:45:42.000-04:00
JIRA: https://jira.spring.io/browse/INT-4307 Shut down the container (to close the connection) when a message-driven endpoint is stopped while the application continues to run.
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -48,6 +48,18 @@ public void setRequestChannel(MessageChannel requestChannel) {
 		this.endpoint.getListener().setRequestChannel(requestChannel);
 	}
 
+	/**
+	 * Set to false to prevent listener container shutdown when the endpoint is stopped.
+	 * Then, if so configured, any cached consumer(s) in the container will remain.
+	 * Otherwise the shared connection and will be closed and the listener invokers shut
+	 * down; this behavior is new starting with version 5.1. Default: true.
+	 * @param shutdownContainerOnStop false to not shutdown.
+	 * @since 5.1
+	 */
+	public void setShutdownContainerOnStop(boolean shutdownContainerOnStop) {
+		this.endpoint.setShutdownContainerOnStop(shutdownContainerOnStop);
+	}
+
 	@Override
 	public String getComponentType() {
 		return this.endpoint.getComponentType();
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.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.
@@ -44,7 +44,11 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements
 
 	private final ChannelPublishingJmsMessageListener listener;
 
-	private volatile String sessionAcknowledgeMode;
+	private String sessionAcknowledgeMode;
+
+	private boolean shutdownContainerOnStop = true;
+
+	private volatile boolean hasStopped;
 
 	/**
 	 * Construct an instance with an externally configured container.
@@ -128,6 +132,18 @@ public void setShouldTrack(boolean shouldTrack) {
 		this.listener.setShouldTrack(shouldTrack);
 	}
 
+	/**
+	 * Set to false to prevent listener container shutdown when the endpoint is stopped.
+	 * Then, if so configured, any cached consumer(s) in the container will remain.
+	 * Otherwise the shared connection and will be closed and the listener invokers shut
+	 * down; this behavior is new starting with version 5.1. Default: true.
+	 * @param shutdownContainerOnStop false to not shutdown.
+	 * @since 5.1
+	 */
+	public void setShutdownContainerOnStop(boolean shutdownContainerOnStop) {
+		this.shutdownContainerOnStop = shutdownContainerOnStop;
+	}
+
 	public ChannelPublishingJmsMessageListener getListener() {
 		return this.listener;
 	}
@@ -177,13 +193,21 @@ protected void onInit() {
 	protected void doStart() {
 		this.listener.start();
 		if (!this.listenerContainer.isRunning()) {
+			if (this.hasStopped) {
+				this.listenerContainer.initialize();
+				this.hasStopped = false;
+			}
 			this.listenerContainer.start();
 		}
 	}
 
 	@Override
 	protected void doStop() {
 		this.listenerContainer.stop();
+		if (this.shutdownContainerOnStop) {
+			this.hasStopped = true;
+			this.listenerContainer.shutdown();
+		}
 		this.listener.stop();
 	}
 
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java
@@ -35,6 +35,7 @@
  * @param <S> the target {@link JmsInboundGatewaySpec} implementation type.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
@@ -176,6 +177,20 @@ public S extractReplyPayload(boolean extractReplyPayload) {
 		return _this();
 	}
 
+	/**
+	 * Set to false to prevent listener container shutdown when the endpoint is stopped.
+	 * Then, if so configured, any cached consumer(s) in the container will remain.
+	 * Otherwise the shared connection and will be closed and the listener invokers shut
+	 * down; this behavior is new starting with version 5.1. Default: true.
+	 * @param shutdown false to not shutdown.
+	 * @return the spec.
+	 * @since 5.1
+	 */
+	public S shutdownContainerOnStop(boolean shutdown) {
+		this.target.setShutdownContainerOnStop(shutdown);
+		return _this();
+	}
+
 	/**
 	 * An {@link AbstractMessageListenerContainer}-based {@link JmsInboundGatewaySpec} extension.
 	 *
diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -37,6 +37,7 @@
  * @param <S> the target {@link JmsMessageDrivenChannelAdapterSpec} implementation type.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
@@ -78,6 +79,20 @@ public S extractPayload(boolean extractRequestPayload) {
 		return _this();
 	}
 
+	/**
+	 * Set to false to prevent listener container shutdown when the endpoint is stopped.
+	 * Then, if so configured, any cached consumer(s) in the container will remain.
+	 * Otherwise the shared connection and will be closed and the listener invokers shut
+	 * down; this behavior is new starting with version 5.1. Default: true.
+	 * @param shutdown false to not shutdown.
+	 * @return the spec.
+	 * @since 5.1
+	 */
+	public S shutdownContainerOnStop(boolean shutdown) {
+		this.target.setShutdownContainerOnStop(shutdown);
+		return _this();
+	}
+
 	/**
 	 *
 	 * @param <S> the target {@link JmsListenerContainerSpec} implementation type.
diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointTests.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.jms.config;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import javax.jms.ConnectionFactory;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
+import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 5.1
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class JmsMessageDrivenEndpointTests {
+
+	@Test
+	public void testStopStart(@Autowired JmsTemplate template,
+			@Autowired JmsMessageDrivenEndpoint endpoint, @Autowired QueueChannel out) {
+		template.convertAndSend("stop.start", "foo");
+		assertThat(out.receive(10_000).getPayload()).isEqualTo("foo");
+		endpoint.stop();
+		assertThat(TestUtils.getPropertyValue(endpoint, "listenerContainer.sharedConnection")).isNull();
+		endpoint.start();
+		template.convertAndSend("stop.start", "bar");
+		assertThat(out.receive(10_000).getPayload()).isEqualTo("bar");
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class Config {
+
+
+		@Bean
+		public ConnectionFactory cf() {
+			return new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
+		}
+
+		@Bean
+		public CachingConnectionFactory ccf() {
+			return new CachingConnectionFactory(cf());
+		}
+
+		@Bean
+		public JmsTemplate template() {
+			return new JmsTemplate(ccf());
+		}
+
+		@Bean
+		public JmsMessageDrivenEndpoint inbound() {
+			JmsMessageDrivenEndpoint endpoint = new JmsMessageDrivenEndpoint(container(), listener());
+			return endpoint;
+		}
+
+		@Bean
+		public AbstractMessageListenerContainer container() {
+			DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
+			container.setConnectionFactory(cf());
+			container.setSessionTransacted(true);
+			container.setDestinationName("stop.start");
+			return container;
+		}
+
+		@Bean
+		public ChannelPublishingJmsMessageListener listener() {
+			ChannelPublishingJmsMessageListener listener = new ChannelPublishingJmsMessageListener();
+			listener.setRequestChannel(out());
+			return listener;
+		}
+
+		@Bean
+		public QueueChannel out() {
+			return new QueueChannel();
+		}
+
+	}
+
+}
diff --git a/src/reference/asciidoc/jms.adoc b/src/reference/asciidoc/jms.adoc
@@ -133,6 +133,10 @@ When consuming from topics, set the `pub-sub-domain` attribute to true.
 Set `subscription-durable` to `true` for a durable subscription or `subscription-shared` for a shared subscription (which requires a JMS 2.0 broker and has been available since version 4.2).
 Use `subscription-name` to name the subscription.
 
+Starting with version 5.1, when the endpoint is stopped while the application remains running, the underlying listener container is shut down, closing its shared connection and consumers.
+Previously, the connection and consumers remained open.
+To revert to the previous behavior, set the `shutdownContainerOnStop` on the `JmsMessageDrivenEndpoint` to `false`.
+
 [[jms-md-conversion-errors]]
 ==== Inbound Conversion Errors
 
@@ -249,6 +253,10 @@ IMPORTANT: Starting with version 4.2, the default `acknowledge` mode is `transac
 In that case, you should configure the container as needed.
 We recommend that you use `transacted` with the `DefaultMessageListenerContainer` to avoid message loss.
 
+Starting with version 5.1, when the endpoint is stopped while the application remains running, the underlying listener container is shut down, closing its shared connection and consumers.
+Previously, the connection and consumers remained open.
+To revert to the previous behavior, set the `shutdownContainerOnStop` on the `JmsInboundGateway` to `false`.
+
 [[jms-outbound-gateway]]
 === Outbound Gateway
 
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc
@@ -161,5 +161,7 @@ See https://github.com/spring-projects/spring-integration-extensions/tree/master
 
 The `JmsSendingMessageHandler` now provides `deliveryModeExpression` and `timeToLiveExpression` options to determine respective QoS options for JMS message to send at runtime.
 The `DefaultJmsHeaderMapper` now allows to map inbound `JMSDeliveryMode` and `JMSExpiration` properties via setting to `true` respective `setMapInboundDeliveryMode()` and `setMapInboundExpiration()` options.
+When a `JmsMessageDrivenEndpoint` or `JmsInboundGateway` is stopped, the associated listener container is now shut down; this closes its shared connection and any consumers.
+You can configure the endpoints to revert to the previous behavior.
 
 See <<jms>> for more information.
