Merge pull request #514 from garyrussell/INT-2515

* garyrussell-INT-2515:
  INT-2515 More Orderly Shutdown

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java

@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
 /**
  * Adapter that receives Messages from an AMQP Queue, converts them into
  * Spring Integration Messages, and sends the results to a Message Channel.
- * 
+ *
  * @author Mark Fisher
  * @author Gary Russell
  * @since 2.1
@@ -90,4 +90,24 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		this.messageListenerContainer.stop();
 	}
 
+
+	/**
+	 * {@inheritDoc}
+	 * <p/>
+	 * Shuts down the listener container.
+	 */
+	public int beforeShutdown() {
+		this.stop();
+		return 0;
+	}
+
+
+	/**
+	 * {@inheritDoc}
+	 * <p/>No-op
+	 */
+	public int afterShutdown() {
+		return 0;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/core/OrderlyShutdownCapable.java

@@ -16,15 +16,33 @@
 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.
+ * Interface for components that wish to be considered for
+ * an orderly shutdown using management interfaces. beforeShuddown()
+ * will be called before schedulers, executors etc, are stopped.
+ * afterShutdown() is called after the shutdown delay.
+ *
  * @author Gary Russell
  * @since 2.2
  *
  */
 public interface OrderlyShutdownCapable {
 
+	/**
+	 * Called before shutdown begins. Implementations should
+	 * stop accepting new messages. Can optionally return the
+	 * number of active messages in process.
+	 * @return The number of active messages if available.
+	 */
+	int beforeShutdown();
+
+	/**
+	 * Called after normal shutdown of schedulers, executors etc,
+	 * and after the shutdown delay has elapsed, but before any
+	 * forced shutdown of any remaining active scheduler/executor
+	 * threads.Can optionally return the number of active messages
+	 * still in process.
+	 * @return The number of active messages if available.
+	 */
+	int afterShutdown();
+
 }

spring-integration-http/src/main/java/org/springframework/integration/http/inbound/HttpRequestHandlingEndpointSupport.java

@@ -21,6 +21,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -49,6 +50,7 @@ import org.springframework.http.server.ServletServerHttpResponse;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.MessagingException;
+import org.springframework.integration.core.OrderlyShutdownCapable;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.integration.http.converter.MultipartAwareFormHttpMessageConverter;
 import org.springframework.integration.http.converter.SerializingHttpMessageConverter;
@@ -95,7 +97,8 @@ import org.springframework.web.util.UrlPathHelper;
  * @author Gary Russell
  * @since 2.0
  */
-abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySupport {
+abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySupport
+	implements OrderlyShutdownCapable {
 
 	private static final boolean jaxb2Present = ClassUtils.isPresent("javax.xml.bind.Binder",
 			HttpRequestHandlingEndpointSupport.class.getClassLoader());
@@ -132,6 +135,10 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor
 
 	private volatile Map<String, Expression> headerExpressions;
 
+	private volatile boolean shuttingDown;
+
+	private final AtomicInteger activeCount = new AtomicInteger();
+
 	public HttpRequestHandlingEndpointSupport() {
 		this(true);
 	}
@@ -264,6 +271,10 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor
 		this.multipartResolver = multipartResolver;
 	}
 
+	protected boolean isShuttingDown() {
+		return this.shuttingDown;
+	}
+
 	@Override
 	public String getComponentType() {
 		return (this.expectReply) ? "http:inbound-gateway" : "http:inbound-channel-adapter";
@@ -298,13 +309,30 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor
 		this.validateSupportedMethods();
 	}
 
+
+	@Override
+	protected void doStart() {
+		this.shuttingDown = false;
+		super.doStart();
+	}
+
 	/**
 	 * Handles the HTTP request by generating a Message and sending it to the request channel. If this gateway's
 	 * 'expectReply' property is true, it will also generate a response from the reply Message once received.
 	 * @return a the response Message
 	 */
-	@SuppressWarnings({ "rawtypes", "unchecked" })
 	protected final Message<?> doHandleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse) throws IOException {
+		if (this.isShuttingDown()) {
+			return createServiceUnavailableResponse();
+		}
+		else {
+			return actualDoHandleRequest(servletRequest, servletResponse);
+		}
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private Message<?> actualDoHandleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse) throws IOException {
+		this.activeCount.incrementAndGet();
 		try {
 			ServletServerHttpRequest request = this.prepareRequest(servletRequest);
 			if (!this.supportedMethods.contains(request.getMethod())) {
@@ -386,9 +414,22 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor
 		}
 		finally {
 			this.postProcessRequest(servletRequest);
+			this.activeCount.decrementAndGet();
 		}
 	}
 
+	/**
+	 * @return
+	 */
+	private Message<?> createServiceUnavailableResponse() {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Endpoint is shutting down; returning status " + HttpStatus.SERVICE_UNAVAILABLE);
+		}
+		return MessageBuilder.withPayload("Endpoint is shutting down")
+				.setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, HttpStatus.SERVICE_UNAVAILABLE)
+				.build();
+	}
+
 	/**
 	 * Converts the reply message to the appropriate HTTP reply object and
 	 * sets up the servlet response.
@@ -521,4 +562,13 @@ abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewaySuppor
 			}
 		}
 	}
+
+	public int beforeShutdown() {
+		this.shuttingDown = true;
+		return this.activeCount.get();
+	}
+
+	public int afterShutdown() {
+		return this.activeCount.get();
+	}
 }

spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java

@@ -23,9 +23,16 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import org.junit.Test;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.http.HttpStatus;
 import org.springframework.integration.Message;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -246,5 +253,69 @@ public class HttpRequestHandlingControllerTests {
 		assertTrue("Wrong message: "+error, ((String)error.getArguments()[1]).startsWith("failed to send Message"));
 	}
 
+	@Test
+	public void shutDown() throws Exception {
+		DirectChannel requestChannel = new DirectChannel();
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(1);
+		AbstractReplyProducingMessageHandler handler = new AbstractReplyProducingMessageHandler() {
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				try {
+					latch2.countDown();
+					// hold up an active thread so we can verify the count and that it completes ok
+					latch1.await(10, TimeUnit.SECONDS);
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+				}
+				return requestMessage.getPayload().toString().toUpperCase();
+			}
+		};
+		requestChannel.subscribe(handler);
+		final HttpRequestHandlingController controller = new HttpRequestHandlingController(true);
+		controller.setRequestChannel(requestChannel);
+		controller.setViewName("foo");
+		final MockHttpServletRequest request = new MockHttpServletRequest();
+		request.setMethod("POST");
+		request.setContent("hello".getBytes());
+		request.setContentType("text/plain");
+		MockHttpServletResponse response = new MockHttpServletResponse();
+		final AtomicInteger active = new AtomicInteger();
+		final AtomicBoolean expected503 = new AtomicBoolean();
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			public void run() {
+				try {
+					// wait for the active thread
+					latch2.await(10, TimeUnit.SECONDS);
+				}
+				catch (InterruptedException e1) {
+					Thread.currentThread().interrupt();
+				}
+				// start the shutdown
+				active.set(controller.beforeShutdown());
+				try {
+					MockHttpServletResponse response = new MockHttpServletResponse();
+					controller.handleRequest(request, response);
+					expected503.set(response.getStatus() == HttpStatus.SERVICE_UNAVAILABLE.value());
+					latch1.countDown();
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+				}
+			}
+		});
+		ModelAndView modelAndView = controller.handleRequest(request, response);
+		// verify we get a 503 after shutdown starts
+		assertEquals(1, active.get());
+		assertTrue(expected503.get());
+		// verify the active request still processed ok
+		assertEquals("foo", modelAndView.getViewName());
+		assertEquals(1, modelAndView.getModel().size());
+		Object reply = modelAndView.getModel().get("reply");
+		assertNotNull(reply);
+		assertEquals("HELLO", reply);
+	}
+
 
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java

@@ -18,8 +18,10 @@ package org.springframework.integration.ip.tcp;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.integration.Message;
+import org.springframework.integration.core.OrderlyShutdownCapable;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -37,7 +39,7 @@ import org.springframework.util.Assert;
 /**
  * Inbound Gateway using a server connection factory - threading is controlled by the
  * factory. For java.net connections, each socket can process only one message at a time.
- * For java.nio connections, messages may be multiplexed but the client will need to 
+ * For java.nio connections, messages may be multiplexed but the client will need to
  * provide correlation logic. If the client is a {@link TcpOutboundGateway} multiplexing
  * is not used, but multiple concurrent connections can be used if the connection factory uses
  * single-use connections. For true asynchronous bi-directional communication, a pair of
@@ -47,7 +49,7 @@ import org.springframework.util.Assert;
  *
  */
 public class TcpInboundGateway extends MessagingGatewaySupport implements
-		TcpListener, TcpSender, ClientModeCapable {
+		TcpListener, TcpSender, ClientModeCapable, OrderlyShutdownCapable {
 
 	private volatile AbstractServerConnectionFactory serverConnectionFactory;
 
@@ -67,7 +69,29 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 
 	private volatile boolean active;
 
+	private volatile boolean shuttingDown;
+
+	private final AtomicInteger activeCount = new AtomicInteger();
+
 	public boolean onMessage(Message<?> message) {
+		if (this.shuttingDown) {
+			if (logger.isInfoEnabled()) {
+				logger.info("Inbound message ignored; shutting down; " + message.toString());
+			}
+		}
+		else {
+			this.activeCount.incrementAndGet();
+			try {
+				return doOnMessage(message);
+			}
+			finally {
+				this.activeCount.decrementAndGet();
+			}
+		}
+		return false;
+	}
+
+	private boolean doOnMessage(Message<?> message) {
 		Message<?> reply = this.sendAndReceiveMessage(message);
 		if (reply == null) {
 			if (logger.isDebugEnabled()) {
@@ -92,7 +116,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 		return false;
 	}
 
-	/** 
+	/**
 	 * @return true if the associated connection factory is listening.
 	 */
 	public boolean isListening() {
@@ -126,6 +150,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 	public void removeDeadConnection(TcpConnection connection) {
 		connections.remove(connection.getConnectionId());
 	}
+	@Override
 	public String getComponentType(){
 		return "ip:tcp-inbound-gateway";
 	}
@@ -146,6 +171,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 		super.doStart();
 		if (!this.active) {
 			this.active = true;
+			this.shuttingDown = false;
 			if (this.serverConnectionFactory != null) {
 				this.serverConnectionFactory.start();
 			}
@@ -243,4 +269,13 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 		}
 	}
 
+	public int beforeShutdown() {
+		this.shuttingDown = true;
+		return this.activeCount.get();
+	}
+
+	public int afterShutdown() {
+		this.stop();
+		return this.activeCount.get();
+	}
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java

@@ -16,8 +16,10 @@
 package org.springframework.integration.ip.tcp;
 
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.integration.Message;
+import org.springframework.integration.core.OrderlyShutdownCapable;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
@@ -31,17 +33,17 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.Assert;
 
 /**
- * Tcp inbound channel adapter using a TcpConnection to 
+ * Tcp inbound channel adapter using a TcpConnection to
  * receive data - if the connection factory is a server
  * factory, this Listener owns the connections. If it is
  * a client factory, the sender owns the connection.
- * 
+ *
  * @author Gary Russell
  * @since 2.0
  *
  */
-public class TcpReceivingChannelAdapter 
-	extends MessageProducerSupport implements TcpListener, ClientModeCapable {
+public class TcpReceivingChannelAdapter
+	extends MessageProducerSupport implements TcpListener, ClientModeCapable, OrderlyShutdownCapable {
 
 	private AbstractConnectionFactory clientConnectionFactory;
 
@@ -59,8 +61,25 @@ public class TcpReceivingChannelAdapter
 
 	private volatile boolean active;
 
+	private volatile boolean shuttingDown;
+
+	private final AtomicInteger activeCount = new AtomicInteger();
+
 	public boolean onMessage(Message<?> message) {
-		sendMessage(message);
+		if (this.shuttingDown) {
+			if (logger.isInfoEnabled()) {
+				logger.info("Inbound message ignored; shutting down; " + message.toString());
+			}
+		}
+		else {
+			this.activeCount.incrementAndGet();
+			try {
+				sendMessage(message);
+			}
+			finally {
+				this.activeCount.decrementAndGet();
+			}
+		}
 		return false;
 	}
 
@@ -80,6 +99,7 @@ public class TcpReceivingChannelAdapter
 		super.doStart();
 		if (!this.active) {
 			this.active = true;
+			this.shuttingDown = false;
 			if (this.serverConnectionFactory != null) {
 				this.serverConnectionFactory.start();
 			}
@@ -117,7 +137,7 @@ public class TcpReceivingChannelAdapter
 	 * Sets the client or server connection factory; for this (an inbound adapter), if
 	 * the factory is a client connection factory, the sockets are owned by a sending
 	 * channel adapter and this adapter is used to receive replies.
-	 * 
+	 *
 	 * @param connectionFactory the connectionFactory to set
 	 */
 	public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
@@ -126,7 +146,7 @@ public class TcpReceivingChannelAdapter
 		} else {
 			this.serverConnectionFactory = connectionFactory;
 		}
-		connectionFactory.registerListener(this);		
+		connectionFactory.registerListener(this);
 	}
 
 	public boolean isListening() {
@@ -139,6 +159,7 @@ public class TcpReceivingChannelAdapter
 		return false;
 	}
 
+	@Override
 	public String getComponentType(){
 		return "ip:tcp-inbound-channel-adapter";
 	}
@@ -221,4 +242,13 @@ public class TcpReceivingChannelAdapter
 		}
 	}
 
+	public int beforeShutdown() {
+		this.shuttingDown = true;
+		return this.activeCount.get();
+	}
+
+	public int afterShutdown() {
+		this.stop();
+		return this.activeCount.get();
+	}
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java

@@ -40,7 +40,6 @@ 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.connection.support.DefaultTcpSocketSupport;
 import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
@@ -54,7 +53,7 @@ import org.springframework.util.Assert;
  *
  */
 public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
-		implements ConnectionFactory, SmartLifecycle, OrderlyShutdownCapable {
+		implements ConnectionFactory, SmartLifecycle {
 
 	protected static final int DEFAULT_REPLY_TIMEOUT = 10000;
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java

@@ -20,6 +20,7 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketException;
 
+import org.springframework.integration.core.OrderlyShutdownCapable;
 import org.springframework.util.Assert;
 
 /**
@@ -31,7 +32,7 @@ import org.springframework.util.Assert;
  * @since 2.0
  */
 public abstract class AbstractServerConnectionFactory
-		extends AbstractConnectionFactory implements Runnable {
+		extends AbstractConnectionFactory implements Runnable, OrderlyShutdownCapable {
 
 	private static final int DEFAULT_BACKLOG = 5;
 
@@ -41,6 +42,8 @@ public abstract class AbstractServerConnectionFactory
 
 	private volatile int backlog = DEFAULT_BACKLOG;
 
+	private volatile boolean shuttingDown;
+
 
 	/**
 	 * The port on which the factory will listen.
@@ -55,6 +58,7 @@ public abstract class AbstractServerConnectionFactory
 		synchronized (this.lifecycleMonitor) {
 			if (!this.isActive()) {
 				this.setActive(true);
+				this.shuttingDown = false;
 				this.getTaskExecutor().execute(this);
 			}
 		}
@@ -85,6 +89,10 @@ public abstract class AbstractServerConnectionFactory
 		return listening;
 	}
 
+	protected boolean isShuttingDown() {
+		return shuttingDown;
+	}
+
 	/**
 	 * Transfers attributes such as (de)serializer, singleUse etc to a new connection.
 	 * For single use sockets, enforces a socket timeout (default 10 seconds).
@@ -167,4 +175,15 @@ public abstract class AbstractServerConnectionFactory
 	public void setPoolSize(int poolSize) {
 		this.setBacklog(poolSize);
 	}
+
+	public int beforeShutdown() {
+		this.shuttingDown = true;
+		return 0;
+	}
+
+	public int afterShutdown() {
+		this.stop();
+		return 0;
+	}
+
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java

@@ -88,15 +88,24 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 					}
 					continue;
 				}
-				if (logger.isDebugEnabled()) {
-					logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
+				if (this.isShuttingDown()) {
+					if (logger.isInfoEnabled()) {
+						logger.info("New connection from " + socket.getInetAddress().getHostAddress()
+								+ " rejected; the server is in the process of shutting down.");
+					}
+					socket.close();
+				}
+				else {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress());
+					}
+					setSocketAttributes(socket);
+					TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost());
+					connection = wrapConnection(connection);
+					this.initializeConnection(connection, socket);
+					this.getTaskExecutor().execute(connection);
+					this.harvestClosedConnections();
 				}
-				setSocketAttributes(socket);
-				TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost());
-				connection = wrapConnection(connection);
-				this.initializeConnection(connection, socket);
-				this.getTaskExecutor().execute(connection);
-				this.harvestClosedConnections();
 			}
 		} catch (Exception e) {
 			// don't log an error if we had a good socket once and now it's closed

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java

@@ -150,17 +150,26 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
 		logger.debug("New accept");
 		SocketChannel channel = server.accept();
-		channel.configureBlocking(false);
-		Socket socket = channel.socket();
-		setSocketAttributes(socket);
-		TcpNioConnection connection = createTcpNioConnection(channel);
-		if (connection == null) {
-			return;
+		if (this.isShuttingDown()) {
+			if (logger.isInfoEnabled()) {
+				logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress()
+						+ " rejected; the server is in the process of shutting down.");
+			}
+			channel.close();
+		}
+		else {
+			channel.configureBlocking(false);
+			Socket socket = channel.socket();
+			setSocketAttributes(socket);
+			TcpNioConnection connection = createTcpNioConnection(channel);
+			if (connection == null) {
+				return;
+			}
+			connection.setTaskExecutor(this.getTaskExecutor());
+			connection.setLastRead(now);
+			this.channelMap.put(channel, connection);
+			channel.register(selector, SelectionKey.OP_READ, connection);
 		}
-		connection.setTaskExecutor(this.getTaskExecutor());
-		connection.setLastRead(now);
-		channelMap.put(channel, connection);
-		channel.register(selector, SelectionKey.OP_READ, connection);
 	}
 
 	private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) {

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

@@ -25,7 +25,7 @@ import org.springframework.util.Assert;
 /**
  * A message-driven endpoint that receive JMS messages, converts them into
  * Spring Integration Messages, and then sends the result to a channel.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
@@ -65,13 +65,15 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements
 		listener.setComponentName(this.getComponentName());
 	}
 
+	@Override
 	protected void doStart() {
 		this.listener.start();
 		if (!this.listenerContainer.isRunning()) {
-			this.listenerContainer.start();	
+			this.listenerContainer.start();
 		}
 	}
 
+	@Override
 	protected void doStop() {
 		this.listenerContainer.stop();
 		this.listener.stop();
@@ -84,4 +86,15 @@ public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements
 		this.listenerContainer.destroy();
 	}
 
+
+	public int beforeShutdown() {
+		this.stop();
+		return 0;
+	}
+
+
+	public int afterShutdown() {
+		return 0;
+	}
+
 }

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java

@@ -37,7 +37,6 @@ import javax.management.modelmbean.ModelMBean;
 import org.aopalliance.aop.Advice;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.aop.Advisor;
 import org.springframework.aop.PointcutAdvisor;
 import org.springframework.aop.TargetSource;
@@ -463,7 +462,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 	 */
 	public void run() {
 		try {
-			this.stopOrderlyShutdownCapableComponents();
+			this.orderlyShutdownCapableComponentsBefore();
 			this.stopActiveChannels();
 			this.stopSchedulers();
 			if (System.currentTimeMillis() > this.shutdownDeadline) {
@@ -487,12 +486,14 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 					Thread.currentThread().interrupt();
 					logger.error("Interrupted while waiting for quiesce");
 				}
+				this.orderlyShutdownCapableComponentsAfter();
 			}
 			else {
 				this.shutdownForced = true;
 				this.stopSchedulers();
 				this.stopExecutors();
 				this.stopNonSpringExecutors();
+				this.orderlyShutdownCapableComponentsAfter();
 			}
 		}
 		finally {
@@ -610,21 +611,32 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 		logger.debug("Stopped other executors");
 	}
 
-	@ManagedOperation
-	public void stopOrderlyShutdownCapableComponents() {
-		logger.debug("Stopping OrderlyShutdownCapable components");
-		Map<String, OrderlyShutdownCapable> candidates = this.applicationContext
+	protected final void orderlyShutdownCapableComponentsBefore() {
+		logger.debug("Initiating stop OrderlyShutdownCapable components");
+		Map<String, OrderlyShutdownCapable> components = this.applicationContext
 				.getBeansOfType(OrderlyShutdownCapable.class);
-		for (Entry<String, OrderlyShutdownCapable> candidateEntry : candidates.entrySet()) {
-			OrderlyShutdownCapable candidate = candidateEntry.getValue();
-			if (candidate instanceof Lifecycle) {
-				if (logger.isInfoEnabled()) {
-					logger.info("Stopping component " + candidate);
-				}
-				((Lifecycle) candidate).stop();
+		for (Entry<String, OrderlyShutdownCapable> componentEntry : components.entrySet()) {
+			OrderlyShutdownCapable component = componentEntry.getValue();
+			int n = component.beforeShutdown();
+			if (logger.isInfoEnabled()) {
+				logger.info("Initiated stop for component " + component + "; it reported " + n + " active messages");
 			}
 		}
-		logger.debug("Stopped OrderlyShutdownCapable components");
+		logger.debug("Initiated stop OrderlyShutdownCapable components");
+	}
+
+	protected final void orderlyShutdownCapableComponentsAfter() {
+		logger.debug("Finalizing stop OrderlyShutdownCapable components");
+		Map<String, OrderlyShutdownCapable> components = this.applicationContext
+				.getBeansOfType(OrderlyShutdownCapable.class);
+		for (Entry<String, OrderlyShutdownCapable> componentEntry : components.entrySet()) {
+			OrderlyShutdownCapable component = componentEntry.getValue();
+			int n = component.afterShutdown();
+			if (logger.isInfoEnabled()) {
+				logger.info("Finalized stop for component " + component + "; it reported " + n + " active messages");
+			}
+		}
+		logger.debug("Initiated stop OrderlyShutdownCapable components");
 	}
 
 	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Channel Count")

spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java

@@ -1,11 +1,11 @@
 /*
  * 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
- * 
+ *
  * 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.
@@ -44,9 +44,9 @@ import org.springframework.util.Assert;
 public class MBeanExporterIntegrationTests {
 
 	private IntegrationMBeanExporter messageChannelsMonitor;
-	
+
 	private GenericXmlApplicationContext context;
-	
+
 	@After
 	public void close() {
 		if (context!=null) {
@@ -55,28 +55,28 @@ public class MBeanExporterIntegrationTests {
 	}
 
 	@Test
-	public void testCircularReferenceNoChannel() throws Exception {		
+	public void testCircularReferenceNoChannel() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "oref-nonchannel.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
 	}
 
 	@Test
-	public void testCircularReferenceNoChannelInFactoryBean() throws Exception {		
+	public void testCircularReferenceNoChannelInFactoryBean() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "oref-factory-nonchannel.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
 	}
 
 	@Test
-	public void testCircularReferenceWithChannel() throws Exception {		
+	public void testCircularReferenceWithChannel() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "oref-channel.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
 	}
 
 	@Test
-	public void testCircularReferenceWithChannelInFactoryBean() throws Exception {		
+	public void testCircularReferenceWithChannelInFactoryBean() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "oref-factory-channel.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
@@ -87,14 +87,14 @@ public class MBeanExporterIntegrationTests {
 	}
 
 	@Test
-	public void testCircularReferenceWithChannelInFactoryBeanAutodetected() throws Exception {		
+	public void testCircularReferenceWithChannelInFactoryBeanAutodetected() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "oref-factory-channel-autodetect.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
 	}
 
 	@Test
-	public void testLifecycleInEndpointWithMessageSource() throws Exception {		
+	public void testLifecycleInEndpointWithMessageSource() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "lifecycle-source.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
@@ -119,7 +119,8 @@ public class MBeanExporterIntegrationTests {
 		ActiveChannel activeChannel = context.getBean("activeChannel", ActiveChannel.class);
 		assertTrue(activeChannel.isStopCalled());
 		OtherActiveComponent otherActiveComponent = context.getBean(OtherActiveComponent.class);
-		assertTrue(otherActiveComponent.isStopCalled());
+		assertTrue(otherActiveComponent.isBeforeCalled());
+		assertTrue(otherActiveComponent.isAfterCalled());
 	}
 
 	@Test
@@ -138,7 +139,7 @@ public class MBeanExporterIntegrationTests {
 	}
 
 	@Test
-	public void testLifecycleInEndpointWithoutMessageSource() throws Exception {		
+	public void testLifecycleInEndpointWithoutMessageSource() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "lifecycle-no-source.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
@@ -160,7 +161,7 @@ public class MBeanExporterIntegrationTests {
 	}
 
 	@Test
-	public void testComponentNames() throws Exception {		
+	public void testComponentNames() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "excluded-components.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
@@ -173,7 +174,7 @@ public class MBeanExporterIntegrationTests {
 	}
 
 	@Test
-	public void testDuplicateComponentNames() throws Exception {		
+	public void testDuplicateComponentNames() throws Exception {
 		context = new GenericXmlApplicationContext(getClass(), "duplicate-components.xml");
 		messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class);
 		assertNotNull(messageChannelsMonitor);
@@ -181,12 +182,12 @@ public class MBeanExporterIntegrationTests {
 		Set<ObjectName> names = server.queryNames(ObjectName.getInstance("org.springframework.integration:type=ManagedEndpoint,*"), null);
 		assertEquals(2, names.size());
 	}
-	
+
 	public static class BogusEndpoint extends AbstractEndpoint {
-		
+
 		@SuppressWarnings("unused")
 		private IntegrationObjectSupport parent;
-		
+
 		public void setParent(IntegrationObjectSupport parent) {
 			this.parent = parent;
 			setComponentName(parent.getComponentName());
@@ -199,7 +200,7 @@ public class MBeanExporterIntegrationTests {
 		@Override
 		protected void doStop() {
 		}
-		
+
 	}
 
 	public static class DateFactoryBean implements FactoryBean<Date> {
@@ -265,16 +266,16 @@ public class MBeanExporterIntegrationTests {
 		}
 
 	}
-	
+
 	@ManagedResource
 	public static class Metric {
-		
+
 	}
 
 	public static class MetricHolder implements InitializingBean {
 
 		private MessageChannel channel;
-		
+
 		public void setChannel(MessageChannel channel) {
 			this.channel = channel;
 		}
@@ -289,7 +290,7 @@ public class MBeanExporterIntegrationTests {
 		String execute() throws Exception;
 		int getCounter();
 	}
-	
+
 	public static class SimpleService implements Service {
 		private int counter;
 
@@ -336,23 +337,28 @@ public class MBeanExporterIntegrationTests {
 		}
 	}
 
-	public static class OtherActiveComponent implements OrderlyShutdownCapable, Lifecycle {
+	public static class OtherActiveComponent implements OrderlyShutdownCapable {
 
-		private boolean stopCalled;
+		private boolean beforeCalled;
 
-		public void start() {
+		private boolean afterCalled;
+
+		public boolean isBeforeCalled() {
+			return this.beforeCalled;
 		}
 
-		public void stop() {
-			this.stopCalled = true;
+		protected boolean isAfterCalled() {
+			return afterCalled;
 		}
 
-		public boolean isRunning() {
-			return false;
+		public int beforeShutdown() {
+			this.beforeCalled = true;
+			return 0;
 		}
 
-		public boolean isStopCalled() {
-			return this.stopCalled;
+		public int afterShutdown() {
+			this.afterCalled = true;
+			return 0;
 		}
 	}
 }

src/reference/docbook/jmx.xml

@@ -421,6 +421,19 @@
 				seconds).
 			</para>
 		</section>
+		<section id="jmx-mbean-shutdown">
+			<title>Orderly Shutdown Managed Operation</title>
+
+			<para>
+				The MBean exporter provides a JMX operation to shut down the application
+				in an orderly manner, intended for use before terminating the JVM.
+			</para>
+			<programlisting language="java"><![CDATA[public void stopActiveComponents(boolean force, long howLong)
+]]></programlisting>
+			<para>
+				Its use and operation are described in <xref linkend="jmx-shutdown"/>.
+			</para>
+		</section>
 	</section>
 </section>
 

src/reference/docbook/shutdown.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<section version="5.0" xml:id="jmx-shutdown"
+         xmlns="http://docbook.org/ns/docbook"
+         xmlns:xlink="http://www.w3.org/1999/xlink"
+         xmlns:ns5="http://www.w3.org/1999/xhtml"
+         xmlns:ns4="http://www.w3.org/1998/Math/MathML"
+         xmlns:ns3="http://www.w3.org/2000/svg"
+         xmlns:ns="http://docbook.org/ns/docbook">
+  <title>Orderly Shutdown</title>
+
+	<para>
+		As described in <xref linkend="jmx-mbean-exporter"/>, the MBean exporter provides a JMX operation
+		<emphasis>stopActiveComponents</emphasis>, which is used to stop the application in an orderly manner. The operation
+		has two parameters, a boolean and a long. The boolean indicates whether attempts will be made
+		to stop (interrupt) active threads; in most cases this will be set to <emphasis>false</emphasis> for orderly
+		shutdown. The long parameter indicates how long (in milliseconds) the operation will wait to allow
+		in-flight messages to complete. The operation works as follows:
+	</para>
+	<para>
+		The first step calls <classname>beforeShutdown()</classname> on all beans that implement
+		<classname>OrderlyShutdownCapable</classname>. This allows such components to prepare for shutdown.
+		Examples of components that implement this interface, and what they do with this call include: JMS and
+		AMQP message-driven adapters stop their listener containers; TCP server connection factories stop
+		accepting new connections (while keeping existing connections open); TCP inbound endpoints drop (log)
+		any new messages received; http inbound endpoints return <emphasis>503 - Service Unavailable</emphasis> for any new
+		requests.
+	</para>
+	<para>
+		The second step stops any active channels, such as JMS- or AMQP-backed channels.
+	</para>
+	<para>
+		The third step stops all <classname>TaskScheduler</classname>s, preventing any new
+		scheduled operations (polling etc).
+	</para>
+	<para>
+		The fourth step stops all <classname>TaskExecutor</classname>s, preventing any new
+		tasks from running.
+	</para>
+	<note>
+		If the shutdown is running from a Spring-managed <classname>TaskExecutor</classname>, shutting down that
+		executor would cause all the timeout time to be consumed by this step, because the thread won't terminate).
+		For this reason, either use a dedicated executor (via the shutdownExecutor property on the MBean exporter),
+		or do not use a Spring-managed executor to invoke this operation.
+	</note>
+	<para>
+		The fifth step stops all <classname>MessageSource</classname>s.
+	</para>
+	<para>
+		The sixth step waits for any remaining time left, as defined by the value of the long parameter passed
+		in to the operation. This is intended to allow any in-flight messages to complete their journeys. It is
+		therefore important to select an appropriate timeout when invoking this operation.
+	</para>
+	<para>
+		The seventh step calls <classname>afterShutdown()</classname> on all OrderlyShutdownCapable components.
+		This allows such components to perform final shutdown tasks (closing all open sockets, for example).
+	</para>
+	<note>
+		If no time is left when we get to step 6, it probably means some thread is hung; in which case, the
+		operation attempts a forced shutdown on all schedulers and executors before exiting.
+	</note>
+</section>

src/reference/docbook/system-management.xml

@@ -8,5 +8,6 @@
   <xi:include href="./message-history.xml"/>
   <xi:include href="./message-store.xml"/>
   <xi:include href="./control-bus.xml"/>
- 
+  <xi:include href="./shutdown.xml"/>
+
 </chapter>

src/reference/docbook/whats-new.xml

@@ -1,37 +1,37 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <chapter xmlns="http://docbook.org/ns/docbook" version="5.0" xml:id="whats-new"
-    xmlns:xi="http://www.w3.org/2001/XInclude"
-    xmlns:xlink="http://www.w3.org/1999/xlink">
-    <title>What's new in Spring Integration 2.2?</title>
-    <para>
-        This chapter provides an overview of the new features and improvements
-        that have been introduced with Spring Integration 2.2 If you are interested
-        in even more detail, please take a look at the Issue Tracker tickets that
-        were resolved as part of the 2.2 development process:
-    </para>
+	xmlns:xi="http://www.w3.org/2001/XInclude"
+	xmlns:xlink="http://www.w3.org/1999/xlink">
+	<title>What's new in Spring Integration 2.2?</title>
+	<para>
+		This chapter provides an overview of the new features and improvements
+		that have been introduced with Spring Integration 2.2 If you are interested
+		in even more detail, please take a look at the Issue Tracker tickets that
+		were resolved as part of the 2.2 development process:
+	</para>
 
-    <section id="2.2-general">
-        <title>General</title>
-        <section id="2.2-spring-31">
-            <title>Spring 3.1</title>
-            <para>
-              Spring Integration now uses Spring 3.1.
-            </para>
-        </section>
-        <section id="2.2-amqp-11">
-            <title>Spring-AMQP 1.1</title>
-            <para>
-              Spring Integration now uses Spring AMQP 1.1. This enables several features
-              to be used within a Spring Integration application, including...
-            </para>
-            <itemizedlist>
-              <listitem>A fixed reply queue for the outbound gateway</listitem>
-              <listitem>HA (mirrored) queues</listitem>
-              <listitem>Publisher Confirms</listitem>
-              <listitem>Returned Messages</listitem>
-              <listitem>Support for Dead Letter Exchanges/Dead Letter Queues</listitem>
-            </itemizedlist>
-        </section>
+	<section id="2.2-general">
+		<title>General</title>
+		<section id="2.2-spring-31">
+			<title>Spring 3.1</title>
+			<para>
+			  Spring Integration now uses Spring 3.1.
+			</para>
+		</section>
+		<section id="2.2-amqp-11">
+			<title>Spring-AMQP 1.1</title>
+			<para>
+			  Spring Integration now uses Spring AMQP 1.1. This enables several features
+			  to be used within a Spring Integration application, including...
+			</para>
+			<itemizedlist>
+				<listitem>A fixed reply queue for the outbound gateway</listitem>
+				<listitem>HA (mirrored) queues</listitem>
+				<listitem>Publisher Confirms</listitem>
+				<listitem>Returned Messages</listitem>
+				<listitem>Support for Dead Letter Exchanges/Dead Letter Queues</listitem>
+			</itemizedlist>
+		</section>
 		<section id="2.2-jdbc-11">
 			<title>JDBC Adapter - Stored Procedures Components</title>
 			<para><emphasis>SpEL Support</emphasis></para>
@@ -51,24 +51,34 @@
 				The Stored Procedure components now provide basic JMX support,
 				exposing some of their properties as MBeans:
 			</para>
-            <itemizedlist>
-              <listitem>Stored Procedure Name</listitem>
-              <listitem>Stored Procedure Name Expression</listitem>
-              <listitem>JdbcCallOperations Cache Statistics</listitem>
-            </itemizedlist>
-        </section>
-        <section id="2.2-tx">
-            <title>Transaction Synchronization</title>
-            <para>
-              When running from a transactional poller,
-              mail inbound adapters can be configured to update the mailbox only
-              if the transaction commits.
-            </para>
-        </section>
-    </section>
+			<itemizedlist>
+				<listitem>Stored Procedure Name</listitem>
+				<listitem>Stored Procedure Name Expression</listitem>
+				<listitem>JdbcCallOperations Cache Statistics</listitem>
+			</itemizedlist>
+		</section>
+		<section id="2.2-tx">
+			<title>Transaction Synchronization</title>
+			<para>
+				When running from a transactional poller,
+				mail inbound adapters can be configured to update the mailbox only
+				if the transaction commits.
+			</para>
+		</section>
+		<section id="2.2-shutdown">
+			<title>Orderly Shutdown</title>
+			<para>
+				A method <classname>stopActiveComponents()</classname> has been
+				added to the IntegrationMBeanExporter. This allows a Spring Integration
+				application to be shut down in an orderly manner, disallowing new inbound
+				messages to certain adapters and waiting for some time to allow in-flight
+				messages to complete.
+			</para>
+		</section>
+	</section>
 
-    <section id="2.2-new-components">
-        <title>New Components</title>
+	<section id="2.2-new-components">
+		<title>New Components</title>
 		<section id="2.2-jpa">
 			<title>JPA Endpoints</title>
 			<para>
@@ -94,9 +104,9 @@
 				For more information please see <xref linkend="jpa"/>
 			</para>
 		</section>
-    </section>
-    <section id="2.2-framework-refactorings">
-        <title>Framework Refactoring</title>
+	</section>
+	<section id="2.2-framework-refactorings">
+		<title>Framework Refactoring</title>
 
-    </section>
+	</section>
 </chapter>