From 0a12a496cc57eb461959ad2856d0ef4df20c86f2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 2 Apr 2012 18:01:35 -0400 Subject: [PATCH] INT-2485 Orderly Shutdown Initial commit - @ManagedOperation on IMBE - can be invoked via JMX, , or getting a reference to the IMBE from the application context. INT-2485 Updates After Review Comments (JIRA) * Shutdown Schedulers first, and wait for them * Add a force parameter, which overrides thread pool shutdown options * Shutdown Sources/Channels after all thread pools have stopped * Mark other components as OrderlyShutdownCapable (e.g. JMS/AMQP Listener containers) and shut them down first * Wait for remaining time to allow for quiescence Also * remove TimeUnit parameter (not JMX-friendly); time limit is now always milliseconds * If thread pools don't stop in time limit, force them down. INT-2485 Handle Self-Destruction Add shutdown-executor to IMBE. When the shutdown was called on a Spring-Managed thread, the shutdown was not clean because we timed out waiting for the current thread to terminate. After that, we force terminated other components. Now, by providing a dedicated Executor for the shutdown process, it is used for the shutdown instead of the current thread. This Executor is *not* shutdown. It is not necessary to provide an Executor if the stopActiveComponents() method is called on some other thread that is not involved in the shutdown. Also adds executor name to logs, when available. INT-2485 Polishing Fix MBean object name collision when running all tests. INT-2485 Enable TCP Shutdown Make TCP connection factories 'OrderlyShutdownCapable' so they are stopped before schedulers/executors in order for them to release any executor threads they are holding. INT-2485 Polishing Didn't need DirectFieldAccessor - scheduler and executor have an accessor for the native ExecutorService. Copyrights INT-2485 Polishing schemaLocation version. INT-2485 PR Review Polishing --- .../inbound/AmqpInboundChannelAdapter.java | 7 +- .../core/OrderlyShutdownCapable.java | 30 ++ .../connection/AbstractConnectionFactory.java | 3 +- .../jms/JmsMessageDrivenEndpoint.java | 7 +- .../jmx/config/MBeanExporterParser.java | 3 +- .../monitor/IntegrationMBeanExporter.java | 287 +++++++++++++++++- .../main/resources/META-INF/spring.schemas | 3 +- .../jmx/config/spring-integration-jmx-2.2.xsd | 202 ++++++++++++ .../MBeanExporterIntegrationTests.java | 81 ++++- .../integration/monitor/lifecycle-source.xml | 22 +- .../monitor/self-destruction-context.xml | 40 +++ 11 files changed, 672 insertions(+), 13 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java create mode 100644 spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.2.xsd create mode 100644 spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml 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 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + +