From 76b1d1df06ea09813130863f505cb413397e62bc Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 13 Sep 2018 14:32:49 -0400 Subject: [PATCH] INT-4307: JMS: Endpoint Stop: Shutdown Container 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. --- .../integration/jms/JmsInboundGateway.java | 14 ++- .../jms/JmsMessageDrivenEndpoint.java | 28 ++++- .../jms/dsl/JmsInboundGatewaySpec.java | 15 +++ .../JmsMessageDrivenChannelAdapterSpec.java | 17 ++- .../config/JmsMessageDrivenEndpointTests.java | 111 ++++++++++++++++++ src/reference/asciidoc/jms.adoc | 8 ++ src/reference/asciidoc/whats-new.adoc | 2 + 7 files changed, 191 insertions(+), 4 deletions(-) create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointTests.java 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 index b1d62208b6..69365af73f 100644 --- 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 class JmsInboundGateway extends MessagingGatewaySupport implements Dispos 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 index 36ad11694e..66ed29326a 100644 --- 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 class JmsMessageDrivenEndpoint extends MessageProducerSupport implements 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,6 +193,10 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements protected void doStart() { this.listener.start(); if (!this.listenerContainer.isRunning()) { + if (this.hasStopped) { + this.listenerContainer.initialize(); + this.hasStopped = false; + } this.listenerContainer.start(); } } @@ -184,6 +204,10 @@ public class JmsMessageDrivenEndpoint extends MessageProducerSupport implements @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 index bb02cb0b3f..73bada869d 100644 --- 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 @@ import org.springframework.util.Assert; * @param the target {@link JmsInboundGatewaySpec} implementation type. * * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ @@ -176,6 +177,20 @@ public class JmsInboundGatewaySpec> 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 index 7c7d29a34c..8d900c71c9 100644 --- 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 @@ import org.springframework.util.Assert; * @param the target {@link JmsMessageDrivenChannelAdapterSpec} implementation type. * * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ @@ -78,6 +79,20 @@ public class JmsMessageDrivenChannelAdapterSpec 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 new file mode 100644 index 0000000000..437d69f0b6 --- /dev/null +++ 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 index 6ff44000b2..c4ba698df6 100644 --- 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 index d86db9de0e..452af27a9e 100644 --- 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 <> for more information.