INT-2515 More Orderly Shutdown

* Add beginShutdown() and endShutdown() to OrderlyShutdownCapable
* JMS/AMQP stop listener containers
* TCP (server side)
** after beginShutdown() disallow new connections, drop (log) new messages
** after endShutdown() close server socket
* HTTP (server side)
** after beginShutdown() disallow any new requests (503 Service Unavailable)

* Docbook updates
** What's new section
** Orderly Shutdown section.

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) {