Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4307: JMS: Endpoint Stop: Shutdown Container

Browse Source 
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.
This commit is contained in:
Gary Russell
2018-09-13 14:32:49 -04:00
committed by Artem Bilan
parent 02c8e0ade2
commit 76b1d1df06
7 changed files with 191 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java
View File

@@ -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();

28
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsMessageDrivenEndpoint.java
View File

@@ -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();
}

15
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java
View File

@@ -35,6 +35,7 @@ import org.springframework.util.Assert;
* @param <S> the target {@link JmsInboundGatewaySpec} implementation type.
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.0
*/
@@ -176,6 +177,20 @@ public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
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.
*

17
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.java
View File

@@ -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 <S> the target {@link JmsMessageDrivenChannelAdapterSpec} implementation type.
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.0
*/
@@ -78,6 +79,20 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
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.

111
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointTests.java Normal file
View File

@@ -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();
}
}
}