diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index 9b56d1d51f..884001244c 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -25,6 +25,7 @@ import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; @@ -34,9 +35,11 @@ import org.springframework.util.Assert; * Spring Integration Messages, and sends the results to a Message Channel. * * @author Mark Fisher + * @author Gary Russell * @since 2.1 */ -public class AmqpInboundChannelAdapter extends MessageProducerSupport { +public class AmqpInboundChannelAdapter extends MessageProducerSupport implements + OrderlyShutdownCapable { private final AbstractMessageListenerContainer messageListenerContainer; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java b/spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java new file mode 100644 index 0000000000..53310ed87c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java @@ -0,0 +1,30 @@ +/* + * Copyright 2002-2012 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.core; + +/** + * Marker interface for components that wish to be considered for + * an orderly shutdown using management interfaces. Components that + * implement this interface will be stopped before schedulers, + * executors etc, in order for them to free up any execution + * threads they may be holding. + * @author Gary Russell + * @since 2.2 + * + */ +public interface OrderlyShutdownCapable { + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index ade6d98ac6..15d94aee98 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -40,6 +40,7 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.MessagingException; import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.util.Assert; @@ -51,7 +52,7 @@ import org.springframework.util.Assert; * */ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport - implements ConnectionFactory, Runnable, SmartLifecycle { + implements ConnectionFactory, Runnable, SmartLifecycle, OrderlyShutdownCapable { protected static final int DEFAULT_REPLY_TIMEOUT = 10000; 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 e8916d5eb7..008278e32f 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-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -17,6 +17,7 @@ package org.springframework.integration.jms; import org.springframework.beans.factory.DisposableBean; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.util.Assert; @@ -27,8 +28,10 @@ import org.springframework.util.Assert; * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ -public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements DisposableBean { +public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements + DisposableBean, OrderlyShutdownCapable { private final AbstractMessageListenerContainer listenerContainer; diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java index 75c9889fa0..c03d788bf3 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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 @@ -60,6 +60,7 @@ public class MBeanExporterParser extends AbstractSingleBeanDefinitionParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-domain"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-name-static-properties"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "managed-components", "componentNamePatterns"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "shutdown-executor"); builder.addPropertyValue("server", mbeanServer); this.registerMBeanExporterHelper(parserContext.getRegistry()); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index 0bfc524837..91c86aac06 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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 @@ -14,11 +14,18 @@ package org.springframework.integration.monitor; import java.lang.reflect.Field; +import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Properties; import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; @@ -44,12 +51,17 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; +import org.springframework.core.task.support.ExecutorServiceAdapter; import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessagingException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.jmx.export.MBeanExporter; @@ -62,6 +74,8 @@ import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.jmx.export.assembler.MetadataMBeanInfoAssembler; import org.springframework.jmx.export.naming.MetadataNamingStrategy; import org.springframework.jmx.support.MetricType; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import org.springframework.util.PatternMatchUtils; import org.springframework.util.ReflectionUtils; @@ -94,7 +108,7 @@ import org.springframework.util.ReflectionUtils; */ @ManagedResource public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware, - BeanClassLoaderAware, SmartLifecycle { + ApplicationContextAware, BeanClassLoaderAware, SmartLifecycle, Runnable { private static final Log logger = LogFactory.getLog(IntegrationMBeanExporter.class); @@ -104,6 +118,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private ListableBeanFactory beanFactory; + private ApplicationContext applicationContext; + private Map anonymousHandlerCounters = new HashMap(); private Map anonymousSourceCounters = new HashMap(); @@ -122,6 +138,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private Map sourcesByName = new HashMap(); + private Map allChannelsByName = new HashMap(); + + private Map allHandlersByName = new HashMap(); + + private Map allSourcesByName = new HashMap(); + private Map beansByEndpointName = new HashMap(); private ClassLoader beanClassLoader; @@ -146,6 +168,14 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private String[] componentNamePatterns = { "*" }; + private volatile Executor shutdownExecutor; + + private volatile long shutdownDeadline; + + private volatile boolean shutdownForced; + + private final AtomicBoolean shuttingDown = new AtomicBoolean(); + public IntegrationMBeanExporter() { super(); // Shouldn't be necessary, but to be on the safe side... @@ -191,6 +221,17 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP this.beanFactory = (ListableBeanFactory) beanFactory; } + public void setApplicationContext(ApplicationContext applicationContext) + throws BeansException { + Assert.notNull(applicationContext, "ApplicationContext may not be null"); + this.applicationContext = applicationContext; + } + + public void setShutdownExecutor(Executor shutdownExecutor) { + Assert.notNull(shutdownExecutor, "Shutdown Executor may not be null"); + this.shutdownExecutor = shutdownExecutor; + } + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { if (bean instanceof Advised) { @@ -409,6 +450,217 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } + /** + * Shutdown active components. If the thread calling this method is + * managed by a Spring-managed executor, you should provide a specific + * dedicated executor via the {@link #setShutdownExecutor(Executor))} + * method. When this is provided, the shutdown will be performed on one + * of its threads, instead of the calling thread; thus avoiding + * the situation where we will wait for the current thread to terminate. + *

It is not necessary to supply this executor service if the + * current thread will not, itself, be shutdown as a result of + * calling this method. + *

Note: The supplied executor service + * will not be shut down. + * + * @param force If true, stop the executors with shutdownNow(), canceling + * running tasks. Overrides any settings on schedulers/executors. When true + * may result in error messages being sent to error channels. + * @param howLong The time to wait in total for all activities to complete + * in milliseconds. + */ + @ManagedOperation + public void stopActiveComponents(boolean force, long howLong) { + if (!this.shuttingDown.compareAndSet(false, true)) { + logger.error("Shutdown already in process"); + return; + } + this.shutdownDeadline = System.currentTimeMillis() + howLong; + this.shutdownForced = force; + if (this.shutdownExecutor == null) { + try { + logger.debug("Running shutdown on current thread"); + this.run(); + } catch (Exception e) { + logger.error("Orderly shutdown failed", e); + } + } + else { + logger.debug("Launching shutdown on another thread"); + this.shutdownExecutor.execute(this); + } + } + + /** + * Perform orderly shutdown - called or executed from + * {@link #stopActiveComponents(boolean, long)}. + */ + public void run() { + try { + this.stopOrderlyShutdownCapableComponents(); + this.stopActiveChannels(); + this.stopSchedulers(); + if (System.currentTimeMillis() > this.shutdownDeadline) { + logger.error("Timed out before waiting for all schedulers to complete"); + } + this.stopExecutors(); + if (System.currentTimeMillis() > this.shutdownDeadline) { + logger.error("Timed out before waiting for all executors to complete"); + } + this.stopNonSpringExecutors(); + if (System.currentTimeMillis() > this.shutdownDeadline) { + logger.error("Timed out before waiting for all non-Spring executors to complete"); + } + this.stopMessageSources(); + // Wait any remaining time for messages to quiesce + long timeLeft = this.shutdownDeadline - System.currentTimeMillis(); + if (timeLeft > 0) { + try { + Thread.sleep(timeLeft); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.error("Interrupted while waiting for quiesce"); + } + } + else { + this.shutdownForced = true; + this.stopSchedulers(); + this.stopExecutors(); + this.stopNonSpringExecutors(); + } + } + finally { + this.shuttingDown.set(false); + } + } + + /** + * Stops all message sources - may cause interrupts. + */ + @ManagedOperation + public void stopMessageSources() { + for (Entry entry : this.allSourcesByName.entrySet()) { + MessageSourceMetrics sourceMetrics = entry.getValue(); + if (sourceMetrics instanceof LifecycleMessageSourceMetrics) { + if (logger.isInfoEnabled()) { + logger.info("Stopping message source " + sourceMetrics); + } + ((LifecycleMessageSourceMetrics) sourceMetrics).stop(); + } + else { + if (logger.isInfoEnabled()) { + logger.info("Message source " + sourceMetrics + " cannot be stopped"); + } + } + } + } + + @ManagedOperation + public void stopActiveChannels() { + // Stop any "active" channels (JMS etc). + for (Entry entry : this.allChannelsByName.entrySet()) { + DirectChannelMetrics metrics = entry.getValue(); + MessageChannel channel = metrics.getMessageChannel(); + if (channel instanceof Lifecycle) { + if (logger.isInfoEnabled()) { + logger.info("Stopping channel " + channel); + } + ((Lifecycle) channel).stop(); + } + } + } + + @ManagedOperation + public void stopSchedulers() { + if (logger.isDebugEnabled()) { + logger.debug("Stopping schedulers " + (this.shutdownForced ? "(force)" : "")); + } + List executorServices = new ArrayList(); + Map schedulers = this.applicationContext + .getBeansOfType(ThreadPoolTaskScheduler.class); + for (Entry entry : schedulers.entrySet()) { + ThreadPoolTaskScheduler scheduler = entry.getValue(); + if (logger.isInfoEnabled()) { + logger.info("Stopping scheduler " + scheduler.getThreadNamePrefix()); + } + ExecutorService executorService = scheduler.getScheduledExecutor(); + executorServices.add(executorService); + doShutdownExecutorService(executorService); + } + waitForExecutors(executorServices); + logger.debug("Stopped schedulers"); + } + + @ManagedOperation + public void stopExecutors() { + if (logger.isDebugEnabled()) { + logger.debug("Stopping executors" + (this.shutdownForced ? "(force)" : "")); + } + List executorServices = new ArrayList(); + Map executors = this.applicationContext + .getBeansOfType(ThreadPoolTaskExecutor.class); + for (Entry entry : executors.entrySet()) { + ThreadPoolTaskExecutor executor = entry.getValue(); + if (executor == this.shutdownExecutor) { + logger.debug("Skipping shutdown of shutdown executor"); + } + else { + if (logger.isInfoEnabled()) { + logger.info("Stopping executor " + executor.getThreadNamePrefix()); + } + ExecutorService executorService = executor.getThreadPoolExecutor(); + executorServices.add(executorService); + doShutdownExecutorService(executorService); + } + } + waitForExecutors(executorServices); + logger.debug("Stopped executors"); + } + + @ManagedOperation + public void stopNonSpringExecutors() { + if (logger.isDebugEnabled()) { + logger.debug("Stopping other executors" + (this.shutdownForced ? "(force)" : "")); + } + List executorServices = new ArrayList(); + Map nonSpringExecutors = this.applicationContext + .getBeansOfType(ExecutorService.class); + for (Entry entry : nonSpringExecutors.entrySet()) { + ExecutorService executorService = entry.getValue(); + if (!(executorService instanceof ExecutorServiceAdapter)) { + if (logger.isInfoEnabled()) { + logger.info("Stopping executor service " + executorService); + } + executorServices.add(executorService); + doShutdownExecutorService(executorService); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("Ignoring ExecutorServiceAdapter"); + } + } + } + waitForExecutors(executorServices); + logger.debug("Stopped other executors"); + } + + @ManagedOperation + public void stopOrderlyShutdownCapableComponents() { + logger.debug("Stopping OrderlyShutdownCapable components"); + Map candidates = this.applicationContext + .getBeansOfType(OrderlyShutdownCapable.class); + for (Entry candidateEntry : candidates.entrySet()) { + OrderlyShutdownCapable candidate = candidateEntry.getValue(); + if (candidate instanceof Lifecycle) { + if (logger.isInfoEnabled()) { + logger.info("Stopping component " + candidate); + } + ((Lifecycle) candidate).stop(); + } + } + logger.debug("Stopped OrderlyShutdownCapable components"); + } + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Channel Count") public int getChannelCount() { return channelsByName.size(); @@ -500,9 +752,37 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } + private void doShutdownExecutorService(ExecutorService executorService) { + if (this.shutdownForced) { + executorService.shutdownNow(); + } + else { + executorService.shutdown(); + } + } + + private void waitForExecutors(List executorServices) { + for (ExecutorService executorService : executorServices) { + try { + if (!executorService.awaitTermination(this.shutdownDeadline + - System.currentTimeMillis(), TimeUnit.MILLISECONDS)) { + logger.error("Executor service " + executorService + " failed to terminate"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.error("Interrupted while shutting down executor service " + executorService); + throw new MessagingException("Interrupted while shutting down", e); + } + if (System.currentTimeMillis() > this.shutdownDeadline) { + logger.error("Timed out before waiting for all executor services"); + } + } + } + private void registerChannels() { for (DirectChannelMetrics monitor : channels) { String name = monitor.getName(); + this.allChannelsByName.put(name, monitor); if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) { continue; } @@ -528,6 +808,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP for (SimpleMessageHandlerMetrics source : handlers) { MessageHandlerMetrics monitor = enhanceHandlerMonitor(source); String name = monitor.getName(); + this.allHandlersByName.put(name, monitor); if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) { continue; } @@ -552,6 +833,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP for (SimpleMessageSourceMetrics source : sources) { MessageSourceMetrics monitor = enhanceSourceMonitor(source); String name = monitor.getName(); + this.allSourcesByName.put(name, monitor); if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) { continue; } @@ -901,5 +1183,4 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP ReflectionUtils.makeAccessible(field); return ReflectionUtils.getField(field, target); } - } diff --git a/spring-integration-jmx/src/main/resources/META-INF/spring.schemas b/spring-integration-jmx/src/main/resources/META-INF/spring.schemas index 8390855dd3..cef6c703ab 100644 --- a/spring-integration-jmx/src/main/resources/META-INF/spring.schemas +++ b/spring-integration-jmx/src/main/resources/META-INF/spring.schemas @@ -1,3 +1,4 @@ http\://www.springframework.org/schema/integration/jmx/spring-integration-jmx-2.0.xsd=org/springframework/integration/jmx/config/spring-integration-jmx-2.0.xsd http\://www.springframework.org/schema/integration/jmx/spring-integration-jmx-2.1.xsd=org/springframework/integration/jmx/config/spring-integration-jmx-2.1.xsd -http\://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd=org/springframework/integration/jmx/config/spring-integration-jmx-2.1.xsd \ No newline at end of file +http\://www.springframework.org/schema/integration/jmx/spring-integration-jmx-2.2.xsd=org/springframework/integration/jmx/config/spring-integration-jmx-2.2.xsd +http\://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd=org/springframework/integration/jmx/config/spring-integration-jmx-2.2.xsd \ No newline at end of file diff --git a/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.2.xsd b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.2.xsd new file mode 100644 index 0000000000..cb4e7badad --- /dev/null +++ b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.2.xsd @@ -0,0 +1,202 @@ + + + + + + + + + + + + + + + Defines an inbound Channel Adapter that polls for JMX attribute values. + + + + + + + + + + + + + + + + + + Defines an outbound Gateway which allows for Message-driven invocation of managed operations that + return values + + + + + + + + + + + + + + + + Defines an outbound Channel Adapter for invoking JMX operations. + + + + + + + + + + + + + + + Defines an inbound Channel Adapter that listens for JMX notifications. + + + + + + + + + + + + + + + + + Defines an outbound Channel Adapter that publishes JMX notifications. + + + + + + + + + + + + + + + Exports Message Channels and Endpoints as MBeans. + + + + + + + + + The domain name for the MBeans exported by this Exporter. + + + + + + + + + + + + Static object properties to be used for this domain. These properties are appended to + the ObjectName of all MBeans registered by this component. + + + + + + + Comma separated list of simple patterns for component names to register (defaults to '*'). + The pattern is applied to all components before they are registered, looking for a match on + the 'name' property of the ObjectName. A MessageChannel and a MessageHandler (for instance) + can share a name because they have a different type, so in that case they would either both + be included or both excluded. + + + + + + + + + + + + An Executor used when shutting down the application using the 'stopActiveComponents()' + method. Only required when invoking the operation on a Spring-managed thread, such as + via a from, say, an error flow. Using this executor avoids the + problem where the shutdown will wait for the current thread to terminate, time out, + and then force-close other components. When a dedicated executor is supplied, + the method will not wait on its threads to complete, and will terminate normally. + It is recommended that the executor used here is dedicated for this purpose and + not used elsewhere. + + + + + + + + + + + + Defines inbound operation invoking type + + + + + + + + + + + + + + Defines outbound operation invoking type + + + + + + + + + + + + + + + + Defines the name of the MBeanServer bean to connect to. + + + + + + \ No newline at end of file diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java index e484acc51a..06232281b2 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2009-2010 the original author or authors. + * Copyright 2009-2012 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 @@ -13,8 +13,10 @@ package org.springframework.integration.monitor; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.util.Arrays; import java.util.Date; @@ -28,10 +30,14 @@ import org.junit.After; import org.junit.Test; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; import org.springframework.context.support.GenericXmlApplicationContext; +import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.integration.core.OrderlyShutdownCapable; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.util.Assert; @@ -107,6 +113,28 @@ public class MBeanExporterIntegrationTests { } // Lifecycle method name assertEquals("start", startName); + assertTrue((Boolean) server.invoke(names.iterator().next(), "isRunning", null, null)); + messageChannelsMonitor.stopActiveComponents(false, 3000); + assertFalse((Boolean) server.invoke(names.iterator().next(), "isRunning", null, null)); + ActiveChannel activeChannel = context.getBean("activeChannel", ActiveChannel.class); + assertTrue(activeChannel.isStopCalled()); + OtherActiveComponent otherActiveComponent = context.getBean(OtherActiveComponent.class); + assertTrue(otherActiveComponent.isStopCalled()); + } + + @Test + public void testSelfDestruction() throws Exception { + context = new GenericXmlApplicationContext(getClass(), "self-destruction-context.xml"); + SourcePollingChannelAdapter adapter = context.getBean(SourcePollingChannelAdapter.class); + adapter.start(); + int n = 0; + while (adapter.isRunning()) { + n += 10; + if (n > 10000) { + fail("Adapter failed to stop"); + } + Thread.sleep(10); + } } @Test @@ -276,4 +304,55 @@ public class MBeanExporterIntegrationTests { } } + public static interface ActiveChannel { + boolean isStopCalled(); + } + + public static class ActiveChannelImpl implements MessageChannel, Lifecycle, ActiveChannel { + + private boolean stopCalled; + + public boolean send(Message message) { + return false; + } + + public boolean send(Message message, long timeout) { + return false; + } + + public void start() { + } + + public void stop() { + this.stopCalled = true; + } + + public boolean isRunning() { + return false; + } + + public boolean isStopCalled() { + return this.stopCalled; + } + } + + public static class OtherActiveComponent implements OrderlyShutdownCapable, Lifecycle { + + private boolean stopCalled; + + public void start() { + } + + public void stop() { + this.stopCalled = true; + } + + public boolean isRunning() { + return false; + } + + public boolean isStopCalled() { + return this.stopCalled; + } + } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml index bade282a59..721e00aa84 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml @@ -1,7 +1,11 @@ - @@ -17,4 +21,18 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml new file mode 100644 index 0000000000..99962e0271 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml @@ -0,0 +1,40 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + +