Merge pull request #456 from garyrussell/INT-2419

* INT-2419:
  INT-2419 Thread Starvation Detection

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java

@@ -25,7 +25,7 @@ import org.w3c.dom.Element;
 
 /**
  * Utility methods and constants for IP adapter parsers.
- * 
+ *
  * @author Gary Russell
  * @since 2.0
  */
@@ -119,13 +119,15 @@ public abstract class IpAdapterParserUtils {
 
 	public static final String SOCKET_FACTORY_SUPPORT = "socket-factory-support";
 
+	public static final String BACKLOG = "backlog";
+
 	private IpAdapterParserUtils() {}
 
 	/**
-	 * Adds a constructor-arg to the provided bean definition builder 
+	 * Adds a constructor-arg to the provided bean definition builder
 	 * with the value of the attribute whose name is provided if that
 	 * attribute is defined in the given element.
-	 * 
+	 *
 	 * @param builder the bean definition builder to be configured
 	 * @param element the XML element where the attribute should be defined
 	 * @param attributeName the name of the attribute whose value will be
@@ -142,7 +144,7 @@ public abstract class IpAdapterParserUtils {
 	/**
 	 * @param element
 	 * @param builder
-	 * @param parserContext 
+	 * @param parserContext
 	 */
 	public static void addHostAndPortToConstructor(Element element,
 			BeanDefinitionBuilder builder, ParserContext parserContext) {
@@ -159,18 +161,18 @@ public abstract class IpAdapterParserUtils {
 	/**
 	 * @param element
 	 * @param builder
-	 * @param parserContext 
+	 * @param parserContext
 	 */
 	public static void addPortToConstructor(Element element,
 			BeanDefinitionBuilder builder, ParserContext parserContext) {
 		String port = IpAdapterParserUtils.getPort(element, parserContext);
 		builder.addConstructorArgValue(port);
 	}
-	
+
 	/**
 	 * Asserts that a port attribute is supplied.
 	 * @param element
-	 * @param parserContext 
+	 * @param parserContext
 	 * @return The value of the attribute.
 	 * @throws BeanCreationException if attribute is not provided.
 	 */
@@ -199,7 +201,7 @@ public abstract class IpAdapterParserUtils {
 	/**
 	 * Sets the common port attributes on the bean being built (timeout, receive buffer size,
 	 * send buffer size).
-	 * @param builder 
+	 * @param builder
 	 * @param element
 	 */
 	static void addCommonSocketOptions(BeanDefinitionBuilder builder, Element element) {

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java

@@ -84,7 +84,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 
 	private volatile boolean singleUse;
 
-	private volatile int poolSize = 5;
+	private volatile int backlog = 5;
 
 	private volatile TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
 
@@ -156,7 +156,6 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 		factory.setLookupHost(this.lookupHost);
 		this.mapper.setApplySequence(this.applySequence);
 		factory.setMapper(this.mapper);
-		factory.setPoolSize(this.poolSize);
 		factory.setSerializer(this.serializer);
 		factory.setSingleUse(this.singleUse);
 		factory.setSoKeepAlive(this.soKeepAlive);
@@ -173,6 +172,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 
 	private void setServerAttributes(AbstractServerConnectionFactory factory) {
 		factory.setLocalAddress(this.localAddress);
+		factory.setBacklog(this.backlog);
 	}
 
 	private TcpSocketFactorySupport obtainSocketFactorySupport() {
@@ -349,9 +349,20 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 	/**
 	 * @param poolSize
 	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setPoolSize(int)
+	 * @deprecated
 	 */
+	@Deprecated
 	public void setPoolSize(int poolSize) {
-		this.poolSize = poolSize;
+		logger.warn("poolSize is deprecated; use backlog instead");
+		this.backlog = poolSize;
+	}
+
+	/**
+	 * @param backlog
+	 * @see AbstractServerConnectionFactory#setBacklog(int)
+	 */
+	public void setBacklog(int backlog) {
+		this.backlog = backlog;
 	}
 
 	/**

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java

@@ -30,51 +30,53 @@ import org.w3c.dom.Element;
  *
  */
 public class TcpConnectionParser extends AbstractBeanDefinitionParser {
-	
+
 	@Override
 	protected AbstractBeanDefinition parseInternal(Element element,
 			ParserContext parserContext) {
 		BeanDefinitionBuilder builder = null;
 		String type = element.getAttribute(IpAdapterParserUtils.TCP_CONNECTION_TYPE);
 		if (!StringUtils.hasText(type)) {
-			parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE + 
+			parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
 					" is required for a tcp connection", element);
 		} else if (!"server".equals(type) && !"client".equals(type)) {
-			parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE + 
+			parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
 					" must be 'client' or 'server' for a TCP Connection Factory", element);
-		}		
+		}
 		builder = BeanDefinitionBuilder.genericBeanDefinition(TcpConnectionFactoryFactoryBean.class);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "type");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.HOST);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.PORT);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.USING_NIO);
 		IpAdapterParserUtils.addCommonSocketOptions(builder, element);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.USING_DIRECT_BUFFERS);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.SO_KEEP_ALIVE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.SO_LINGER);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.SO_TCP_NODELAY);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.SO_TRAFFIC_CLASS);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.POOL_SIZE);
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.BACKLOG);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.TASK_EXECUTOR);
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.SERIALIZER);
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.DESERIALIZER);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.SINGLE_USE);
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, 
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.INTERCEPTOR_FACTORY_CHAIN);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.LOOKUP_HOST);

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

@@ -92,8 +92,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 
 	private volatile boolean singleUse;
 
-	private volatile int poolSize = 5;
-
 	private volatile boolean active;
 
 	private volatile TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
@@ -296,10 +294,14 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 	}
 
 	/**
+	 * @deprecated This property is no longer used. If you wish
+	 * to use a fixed thread pool, provide your own Executor
+	 * in {@link #setTaskExecutor(Executor)}.
 	 * @return the poolSize
 	 */
+	@Deprecated
 	public int getPoolSize() {
-		return poolSize;
+		return 0;
 	}
 
 	/**
@@ -372,8 +374,14 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 	}
 
 
+	/**
+	 * @deprecated Default task executor is now a cached rather
+	 * than a fixed pool executor. To use a pool, supply an
+	 * appropriate Executor in {@link AbstractConnectionFactory#setTaskExecutor(Executor)}.
+	 * Use {@link AbstractServerConnectionFactory#setBacklog(int)} to set the connection backlog.
+	 */
+	@Deprecated
 	public void setPoolSize(int poolSize) {
-		this.poolSize = poolSize;
 	}
 
 	public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) {
@@ -438,7 +446,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 			}
 			if (this.taskExecutor == null) {
 				this.privateExecutor = true;
-				this.taskExecutor = Executors.newFixedThreadPool(this.poolSize);
+				this.taskExecutor = Executors.newCachedThreadPool();
 			}
 			return this.taskExecutor;
 		}

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

@@ -20,6 +20,8 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketException;
 
+import org.springframework.util.Assert;
+
 /**
  * Base class for all server connection factories. Server connection factories
  * listen on a port for incoming connections and create new TcpConnection objects
@@ -30,9 +32,13 @@ import java.net.SocketException;
  */
 public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory {
 
-	private boolean listening;
+	private static final int DEFAULT_BACKLOG = 5;
 
-	private String localAddress;
+	private volatile boolean listening;
+
+	private volatile String localAddress;
+
+	private volatile int backlog = DEFAULT_BACKLOG;
 
 
 	/**
@@ -120,4 +126,34 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
 	public void setLocalAddress(String localAddress) {
 		this.localAddress = localAddress;
 	}
+
+
+	/**
+	 * The number of sockets in the server connection backlog.
+	 * @return The backlog.
+	 */
+	public int getBacklog() {
+		return backlog;
+	}
+
+	/**
+	 * The number of sockets in the connection backlog. Default 5;
+	 * increase if you expect high connection rates.
+	 * @param backlog
+	 */
+	public void setBacklog(int backlog) {
+		Assert.isTrue(backlog >= 0, "You cannot set backlog negative");
+		this.backlog = backlog;
+	}
+
+	/**
+	 * @deprecated Default task executor is now a cached rather
+	 * than a fixed pool executor.
+	 * Use {@link #setBacklog(int)} to set the connection backlog.
+	 */
+	@Override
+	@Deprecated
+	public void setPoolSize(int poolSize) {
+		this.setBacklog(poolSize);
+	}
 }

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

@@ -52,8 +52,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 
 	/**
 	 * If no listener registers, exits.
-	 * Accepts incoming connections and creates TcpConnections for each new connection. 
-	 * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the 
+	 * Accepts incoming connections and creates TcpConnections for each new connection.
+	 * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the
 	 * connection {@link TcpConnection#run()} using the task executor.
 	 * I/O errors on the server socket/channel are logged and the factory is stopped.
 	 */
@@ -65,10 +65,10 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		}
 		try {
 			if (this.getLocalAddress() == null) {
-				theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), null);
+				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), null);
 			} else {
 				InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
-				theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), whichNic);
+				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), whichNic);
 			}
 			this.getTcpSocketSupport().postProcessServerSocket(theServerSocket);
 			this.serverSocket = theServerSocket;
@@ -132,6 +132,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		}
 	}
 
+	@Override
 	public void close() {
 		if (this.serverSocket == null) {
 			return;
@@ -158,5 +159,5 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null");
 		this.tcpSocketFactorySupport = tcpSocketFactorySupport;
 	}
-	
+
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.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.
@@ -26,12 +26,15 @@ import java.nio.channels.ClosedChannelException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessagingException;
 import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
 import org.springframework.util.Assert;
 
@@ -44,6 +47,8 @@ import org.springframework.util.Assert;
  */
 public class TcpNioConnection extends AbstractTcpConnection {
 
+	private static final long DEFAULT_PIPE_TIMEOUT = 60000;
+
 	private final SocketChannel socketChannel;
 
 	private final ChannelOutputStream channelOutputStream;
@@ -66,6 +71,8 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	private volatile boolean writingToPipe;
 
+	private volatile long pipeTimeout = DEFAULT_PIPE_TIMEOUT;
+
 	/**
 	 * Constructs a TcpNetConnection for the SocketChannel.
 	 * @param socketChannel the socketChannel
@@ -80,6 +87,10 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		this.channelOutputStream = new ChannelOutputStream();
 	}
 
+	public void setPipeTimeout(long pipeTimeout) {
+		this.pipeTimeout = pipeTimeout;
+	}
+
 	@Override
 	public void close() {
 		doClose();
@@ -265,7 +276,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		this.writingToPipe = true;
 		try {
 			if (this.taskExecutor == null) {
-				this.taskExecutor = Executors.newSingleThreadExecutor();
+				this.taskExecutor = Executors.newCachedThreadPool();
 			}
 			// If there is no assembler running, start one
 			checkForAssembler();
@@ -287,7 +298,28 @@ public class TcpNioConnection extends AbstractTcpConnection {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
 			}
-			this.sendToPipe(this.rawBuffer);
+			final CountDownLatch latch = new CountDownLatch(1);
+			/*
+			 * If there are insufficient threads, either to run the
+			 * write to the pipe, or to assemble the data, we need
+			 * avoid a deadlock (block on the write to the pipe).
+			 * Hence the count down latch.
+			 */
+			this.taskExecutor.execute(new Runnable() {
+				public void run() {
+					try {
+						TcpNioConnection.this.sendToPipe(rawBuffer);
+						latch.countDown();
+					}
+					catch (Exception e) {
+						logger.error(getConnectionId() + " Failed to write to pipe", e);
+					}
+				}
+			});
+			if (!latch.await(this.pipeTimeout , TimeUnit.MILLISECONDS)) {
+				throw new MessagingException("Timed out writing to pipe, probably due to insufficient threads in " +
+						"a fixed thread pool; consider increasing this task executor pool size");
+			}
 		} finally {
 			this.writingToPipe = false;
 		}
@@ -295,6 +327,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	protected void sendToPipe(ByteBuffer rawBuffer) throws IOException {
 		Assert.notNull(rawBuffer, "rawBuffer cannot be null");
+		if (logger.isTraceEnabled()) {
+			logger.trace("Sending " + rawBuffer.limit() + " to pipe");
+		}
 		this.pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());
 		this.pipedOutputStream.flush();
 		rawBuffer.clear();

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

@@ -64,8 +64,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 
 	/**
 	 * If no listener registers, exits.
-	 * Accepts incoming connections and creates TcpConnections for each new connection. 
-	 * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the 
+	 * Accepts incoming connections and creates TcpConnections for each new connection.
+	 * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the
 	 * connection {@link TcpConnection#run()} using the task executor.
 	 * I/O errors on the server socket/channel are logged and the factory is stopped.
 	 */
@@ -84,11 +84,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 			this.serverChannel.configureBlocking(false);
 			if (this.getLocalAddress() == null) {
 				this.serverChannel.socket().bind(new InetSocketAddress(port),
-					Math.abs(this.getPoolSize()));
+					Math.abs(this.getBacklog()));
 			} else {
 				InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
 				this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port),
-						Math.abs(this.getPoolSize()));
+						Math.abs(this.getBacklog()));
 			}
 			final Selector selector = Selector.open();
 			this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
@@ -109,11 +109,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 	/**
 	 * Listens for incoming connections and for notifications that a connected
 	 * socket is ready for reading.
-	 * Accepts incoming connections, registers the new socket with the 
+	 * Accepts incoming connections, registers the new socket with the
 	 * selector for reading.
 	 * When a socket is ready for reading, unregisters the read interest and
 	 * schedules a call to doRead which reads all available data. When the read
-	 * is complete, the socket is again registered for read interest. 
+	 * is complete, the socket is again registered for read interest.
 	 * @param server
 	 * @param selector
 	 * @throws IOException
@@ -176,6 +176,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 		}
 	}
 
+	@Override
 	public void close() {
 		if (this.selector != null) {
 			this.selector.wakeup();

spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd

@@ -456,13 +456,29 @@ apply to TCP outbound adapters and gateways.
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="task-executor" type="xsd:string" />
+			<xsd:attribute name="task-executor" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+A task executor for managing connections; if not specified a
+cached thread pool task executor is used.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="pool-size" type="xsd:string" >
 				<xsd:annotation>
 					<xsd:documentation>
-The number of threads that will be used for socket/channel handling. Only applies
-if an external task-executor is NOT being used. When using an external task executor,
-its configuration specifies the number of threads.
+Deprecated since 2.2; previously it specified the thread pool size if
+an external task executor was not provided; it was also used to set
+the connection backlog for server factories. That setting is now
+specified by the backlog attribute.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="backlog" type="xsd:string" >
+				<xsd:annotation>
+					<xsd:documentation>
+Specifies the connection backlog for server sockets. Does not
+apply to client factories.
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>

spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml

@@ -206,7 +206,6 @@
 		using-nio="#{props['use.nio']}"
 		single-use="true"
 		task-executor="externalTE"
-		pool-size="321"
 		using-direct-buffers="true"
 		interceptor-factory-chain="interceptors"
 	/>
@@ -232,11 +231,18 @@
 		using-nio="true"
 		single-use="true"
 		task-executor="externalTE"
-		pool-size="123"
+		backlog="123"
 		using-direct-buffers="true"
 		interceptor-factory-chain="interceptors"
 	/>
 
+	<ip:tcp-connection-factory
+		id="serverBackwardsCompatible"
+		type="server"
+		port="#{client1.port}"
+		pool-size="123"
+	/>
+
 	<ip:tcp-connection-factory
 		id="client2"
 		type="client"
@@ -254,7 +260,6 @@
 		using-nio="false"
 		single-use="true"
 		task-executor="externalTE"
-		pool-size="321"
 		interceptor-factory-chain="interceptors"
 	/>
 
@@ -275,7 +280,7 @@
 		using-nio="false"
 		single-use="true"
 		task-executor="externalTE"
-		pool-size="123"
+		backlog="123"
 		interceptor-factory-chain="interceptors"
 	/>
 

spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java

@@ -154,6 +154,9 @@ public class ParserUnitTests {
 	@Autowired
 	AbstractConnectionFactory server1;
 
+	@Autowired
+	AbstractConnectionFactory serverBackwardsCompatible;
+
 	@Autowired
 	AbstractConnectionFactory server2;
 
@@ -404,7 +407,6 @@ public class ParserUnitTests {
 		assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
 		assertEquals(true, dfa.getPropertyValue("singleUse"));
 		assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
-		assertEquals(321, dfa.getPropertyValue("poolSize"));
 		assertEquals(true, dfa.getPropertyValue("usingDirectBuffers"));
 		assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
 	}
@@ -424,11 +426,18 @@ public class ParserUnitTests {
 		assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
 		assertEquals(true, dfa.getPropertyValue("singleUse"));
 		assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
-		assertEquals(123, dfa.getPropertyValue("poolSize"));
+		assertEquals(123, dfa.getPropertyValue("backlog"));
 		assertEquals(true, dfa.getPropertyValue("usingDirectBuffers"));
 		assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
 	}
 
+	@Test
+	public void testConnDeprecatedPoolSize() {
+		assertTrue(serverBackwardsCompatible instanceof TcpNetServerConnectionFactory);
+		DirectFieldAccessor dfa = new DirectFieldAccessor(serverBackwardsCompatible);
+		assertEquals(123, dfa.getPropertyValue("backlog"));
+	}
+
 	@Test
 	public void testConnClient2() {
 		assertTrue(client2 instanceof TcpNetClientConnectionFactory);
@@ -445,7 +454,6 @@ public class ParserUnitTests {
 		assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
 		assertEquals(true, dfa.getPropertyValue("singleUse"));
 		assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
-		assertEquals(321, dfa.getPropertyValue("poolSize"));
 		assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
 	}
 
@@ -464,7 +472,7 @@ public class ParserUnitTests {
 		assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
 		assertEquals(true, dfa.getPropertyValue("singleUse"));
 		assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
-		assertEquals(123, dfa.getPropertyValue("poolSize"));
+		assertEquals(123, dfa.getPropertyValue("backlog"));
 		assertNotNull(dfa.getPropertyValue("interceptorFactoryChain"));
 	}
 

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java

@@ -37,7 +37,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import javax.net.ServerSocketFactory;
 
 import org.junit.Test;
-
 import org.springframework.core.serializer.DefaultDeserializer;
 import org.springframework.core.serializer.DefaultSerializer;
 import org.springframework.integration.Message;
@@ -54,7 +53,7 @@ import org.springframework.integration.support.MessageBuilder;
  */
 public class TcpOutboundGatewayTests {
 
-	@Test 
+	@Test
 	public void testGoodNetSingle() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
@@ -84,7 +83,6 @@ public class TcpOutboundGatewayTests {
 		ccf.setDeserializer(new DefaultDeserializer());
 		ccf.setSoTimeout(10000);
 		ccf.setSingleUse(true);
-		ccf.setPoolSize(10);
 		ccf.start();
 		assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
 		TcpOutboundGateway gateway = new TcpOutboundGateway();
@@ -98,7 +96,7 @@ public class TcpOutboundGatewayTests {
 			gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build());
 		}
 		Set<String> replies = new HashSet<String>();
-		for (int i = 100; i < 200; i++) { 
+		for (int i = 100; i < 200; i++) {
 			Message<?> m = replyChannel.receive(10000);
 			assertNotNull(m);
 			replies.add((String) m.getPayload());
@@ -108,7 +106,7 @@ public class TcpOutboundGatewayTests {
 		}
 	}
 
-	@Test 
+	@Test
 	public void testGoodNetMultiplex() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
 		final CountDownLatch latch = new CountDownLatch(1);
@@ -149,7 +147,7 @@ public class TcpOutboundGatewayTests {
 			gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build());
 		}
 		Set<String> replies = new HashSet<String>();
-		for (int i = 100; i < 110; i++) { 
+		for (int i = 100; i < 110; i++) {
 			Message<?> m = replyChannel.receive(10000);
 			assertNotNull(m);
 			replies.add((String) m.getPayload());
@@ -160,7 +158,7 @@ public class TcpOutboundGatewayTests {
 		done.set(true);
 	}
 
-	@Test 
+	@Test
 	public void testGoodNetTimeout() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
 		final CountDownLatch latch = new CountDownLatch(1);
@@ -212,11 +210,11 @@ public class TcpOutboundGatewayTests {
 		}
 		Set<String> replies = new HashSet<String>();
 		int timeouts = 0;
-		for (int i = 0; i < 2; i++) { 
+		for (int i = 0; i < 2; i++) {
 			try {
 				results[i].get();
 			} catch (InterruptedException e) {
-				
+
 			} catch (ExecutionException e) {
 				if (timeouts > 0) {
 					fail("Unexpected " + e.getMessage());

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java

@@ -151,7 +151,7 @@ public class TcpReceivingChannelAdapterTests {
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		scf.setSerializer(serializer);
-		scf.setDeserializer(serializer);		
+		scf.setDeserializer(serializer);
 		scf.setSoTimeout(5000);
 		TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
 		adapter.setConnectionFactory(scf);
@@ -168,7 +168,7 @@ public class TcpReceivingChannelAdapterTests {
 		Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
 		for (int i = 0; i < 1000; i++) {
 			socket.getOutputStream().write(("Test" + i + "\r\n").getBytes());
-		}	
+		}
 		Set<String> results = new HashSet<String>();
 		for (int i = 0; i < 1000; i++) {
 			Message<?> message = channel.receive(10000);
@@ -210,14 +210,14 @@ public class TcpReceivingChannelAdapterTests {
 		handler.handleMessage(message);
 		message = channel.receive(10000);
 		assertNotNull(message);
-		handler.handleMessage(message);		
+		handler.handleMessage(message);
 		byte[] b = new byte[6];
 		readFully(socket.getInputStream(), b);
 		assertEquals("Test\r\n", new String(b));
 		readFully(socket.getInputStream(), b);
 		assertEquals("Test\r\n", new String(b));
 	}
-	
+
 	@Test
 	public void testNioShared() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
@@ -248,14 +248,14 @@ public class TcpReceivingChannelAdapterTests {
 		handler.handleMessage(message);
 		message = channel.receive(10000);
 		assertNotNull(message);
-		handler.handleMessage(message);		
+		handler.handleMessage(message);
 		byte[] b = new byte[6];
 		readFully(socket.getInputStream(), b);
 		assertEquals("Test\r\n", new String(b));
 		readFully(socket.getInputStream(), b);
 		assertEquals("Test\r\n", new String(b));
 	}
-	
+
 	@Test
 	public void testNetSingleNoOutbound() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
@@ -316,7 +316,7 @@ public class TcpReceivingChannelAdapterTests {
 		socket.getOutputStream().write("Test1\r\n".getBytes());
 		socket = SocketFactory.getDefault().createSocket("localhost", port);
 		socket.getOutputStream().write("Test2\r\n".getBytes());
-		Message<?> message = channel.receive(10000);
+		Message<?> message = channel.receive(60000);
 		assertNotNull(message);
 		// with single use, results may come back in a different order
 		Set<String> results = new HashSet<String>();
@@ -371,15 +371,15 @@ public class TcpReceivingChannelAdapterTests {
 		handler.handleMessage(message);
 		message = channel.receive(10000);
 		assertNotNull(message);
-		handler.handleMessage(message);		
+		handler.handleMessage(message);
 		byte[] b = new byte[7];
 		readFully(socket1.getInputStream(), b);
 		assertEquals("Test1\r\n", new String(b));
 		readFully(socket2.getInputStream(), b);
 		assertEquals("Test2\r\n", new String(b));
 	}
-	
-	@Test  
+
+	@Test
 	public void testNioSingleShared() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
@@ -412,15 +412,15 @@ public class TcpReceivingChannelAdapterTests {
 		handler.handleMessage(message);
 		message = channel.receive(10000);
 		assertNotNull(message);
-		handler.handleMessage(message);		
+		handler.handleMessage(message);
 		byte[] b = new byte[7];
 		readFully(socket1.getInputStream(), b);
 		assertEquals("Test1\r\n", new String(b));
 		readFully(socket2.getInputStream(), b);
 		assertEquals("Test2\r\n", new String(b));
 	}
-	
-	@Test  
+
+	@Test
 	public void testNioSingleSharedMany() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
 		TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
@@ -428,12 +428,12 @@ public class TcpReceivingChannelAdapterTests {
 		scf.setSerializer(serializer);
 		scf.setDeserializer(serializer);
 		scf.setSingleUse(true);
-		scf.setPoolSize(100);
+		scf.setBacklog(100);
 		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
 		handler.setConnectionFactory(scf);
 		TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
 		adapter.setConnectionFactory(scf);
-		Executor te = Executors.newFixedThreadPool(10);
+		Executor te = Executors.newCachedThreadPool();
 		scf.setTaskExecutor(te);
 		scf.start();
 		QueueChannel channel = new QueueChannel();
@@ -453,17 +453,17 @@ public class TcpReceivingChannelAdapterTests {
 			sockets.add(socket1);
 		}
 		for (int i = 100; i < 200; i++) {
-			Message<?> message = channel.receive(10000);
+			Message<?> message = channel.receive(60000);
 			assertNotNull(message);
 			handler.handleMessage(message);
 		}
 		byte[] b = new byte[9];
-		for (int i = 100; i < 200; i++) { 
+		for (int i = 100; i < 200; i++) {
 			readFully(sockets.remove(0).getInputStream(), b);
 			assertEquals("Test" + i + "\r\n", new String(b));
 		}
 	}
-	
+
 	@Test
 	public void testNetInterceptors() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
@@ -484,7 +484,7 @@ public class TcpReceivingChannelAdapterTests {
 		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
 		singleSharedInterceptorsGuts(port, scf);
 	}
-	
+
 	@Test
 	public void testNioInterceptors() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
@@ -509,11 +509,11 @@ public class TcpReceivingChannelAdapterTests {
 	private void interceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception {
 		scf.setSerializer(new DefaultSerializer());
 		scf.setDeserializer(new DefaultDeserializer());
-		scf.setSingleUse(false);		
+		scf.setSingleUse(false);
 		TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
 		adapter.setConnectionFactory(scf);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[] 
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
             		     {new HelloWorldInterceptorFactory(),
                		      new HelloWorldInterceptorFactory()});
 		scf.setInterceptorFactoryChain(fc);
@@ -553,7 +553,7 @@ public class TcpReceivingChannelAdapterTests {
 		scf.setSingleUse(true);
 		scf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[] 
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
             		     {new HelloWorldInterceptorFactory(),
                		      new HelloWorldInterceptorFactory()});
 		scf.setInterceptorFactoryChain(fc);
@@ -576,7 +576,7 @@ public class TcpReceivingChannelAdapterTests {
 		new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello");
 		assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject());
 		new ObjectOutputStream(socket.getOutputStream()).writeObject("Test1");
-		
+
 		socket = SocketFactory.getDefault().createSocket("localhost", port);
 		new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello");
 		assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject());
@@ -601,7 +601,7 @@ public class TcpReceivingChannelAdapterTests {
 		scf.setSingleUse(true);
 		scf.setSoTimeout(60000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[] 
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
             		     {new HelloWorldInterceptorFactory(),
                		      new HelloWorldInterceptorFactory()});
 		scf.setInterceptorFactoryChain(fc);
@@ -641,7 +641,7 @@ public class TcpReceivingChannelAdapterTests {
 		message = channel.receive(10000);
 		assertNotNull(message);
 		handler.handleMessage(message);
-		
+
 		assertEquals("Test1", new ObjectInputStream(socket1.getInputStream()).readObject());
 		assertEquals("Test2", new ObjectInputStream(socket2.getInputStream()).readObject());
 	}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java

@@ -51,7 +51,6 @@ import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.PollableChannel;
-import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory;
 import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory;
@@ -80,7 +79,7 @@ public class TcpSendingMessageHandlerTests {
 			buff[i] = (byte) is.read();
 		}
 	}
-	
+
 	@Test
 	public void testNetCrLf() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
@@ -111,7 +110,7 @@ public class TcpSendingMessageHandlerTests {
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		ccf.setSerializer(serializer);
 		ccf.setDeserializer(serializer);
-		ccf.setSoTimeout(10000);		
+		ccf.setSoTimeout(10000);
 		ccf.start();
 		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
 		handler.setConnectionFactory(ccf);
@@ -119,7 +118,7 @@ public class TcpSendingMessageHandlerTests {
 		adapter.setConnectionFactory(ccf);
 		QueueChannel channel = new QueueChannel();
 		adapter.setOutputChannel(channel);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		Message<?> mOut = channel.receive(10000);
@@ -239,7 +238,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(results.remove("Reply2"));
 		done.set(true);
 	}
-	
+
 	@Test
 	public void testNetStxEtx() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
@@ -277,7 +276,7 @@ public class TcpSendingMessageHandlerTests {
 		adapter.setConnectionFactory(ccf);
 		QueueChannel channel = new QueueChannel();
 		adapter.setOutputChannel(channel);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		Message<?> mOut = channel.receive(10000);
@@ -381,7 +380,7 @@ public class TcpSendingMessageHandlerTests {
 		adapter.setConnectionFactory(ccf);
 		QueueChannel channel = new QueueChannel();
 		adapter.setOutputChannel(channel);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		Message<?> mOut = channel.receive(10000);
@@ -484,7 +483,7 @@ public class TcpSendingMessageHandlerTests {
 		adapter.setConnectionFactory(ccf);
 		QueueChannel channel = new QueueChannel();
 		adapter.setOutputChannel(channel);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		Message<?> mOut = channel.receive(10000);
@@ -679,7 +678,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
-		assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS));		
+		assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS));
 		Set<String> replies = new HashSet<String>();
 		for (int i = 0; i < 2; i++) {
 			Message<?> mOut = channel.receive(10000);
@@ -736,7 +735,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
-		assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS));		
+		assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS));
 		Set<String> replies = new HashSet<String>();
 		for (int i = 0; i < 2; i++) {
 			Message<?> mOut = channel.receive(10000);
@@ -859,9 +858,9 @@ public class TcpSendingMessageHandlerTests {
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
-		ccf.setSoTimeout(10000);		
+		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[] 
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
 		     {new HelloWorldInterceptorFactory(),
 		      new HelloWorldInterceptorFactory()});
 		ccf.setInterceptorFactoryChain(fc);
@@ -872,7 +871,7 @@ public class TcpSendingMessageHandlerTests {
 		adapter.setConnectionFactory(ccf);
 		QueueChannel channel = new QueueChannel();
 		adapter.setOutputChannel(channel);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		Message<?> mOut = channel.receive(10000);
@@ -921,7 +920,7 @@ public class TcpSendingMessageHandlerTests {
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
-		ccf.setSoTimeout(10000);		
+		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
 		fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory()});
 		ccf.setInterceptorFactoryChain(fc);
@@ -932,7 +931,7 @@ public class TcpSendingMessageHandlerTests {
 		adapter.setConnectionFactory(ccf);
 		QueueChannel channel = new QueueChannel();
 		adapter.setOutputChannel(channel);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		for (int i = 0; i < 1000; i++) {
 			handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		}
@@ -991,9 +990,9 @@ public class TcpSendingMessageHandlerTests {
 		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
-		ccf.setSoTimeout(10000);		
+		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[] 
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
             		     {new HelloWorldInterceptorFactory(),
                		      new HelloWorldInterceptorFactory()});
 		ccf.setInterceptorFactoryChain(fc);
@@ -1001,7 +1000,7 @@ public class TcpSendingMessageHandlerTests {
 		ccf.start();
 		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
 		handler.setConnectionFactory(ccf);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		done.set(true);
 	}
@@ -1047,9 +1046,9 @@ public class TcpSendingMessageHandlerTests {
 		AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
 		ccf.setSerializer(new DefaultSerializer());
 		ccf.setDeserializer(new DefaultDeserializer());
-		ccf.setSoTimeout(10000);		
+		ccf.setSoTimeout(10000);
 		TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
-		fc.setInterceptors(new TcpConnectionInterceptorFactory[] 
+		fc.setInterceptors(new TcpConnectionInterceptorFactory[]
             		     {new HelloWorldInterceptorFactory(),
                		      new HelloWorldInterceptorFactory()});
 		ccf.setInterceptorFactoryChain(fc);
@@ -1057,7 +1056,7 @@ public class TcpSendingMessageHandlerTests {
 		ccf.start();
 		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
 		handler.setConnectionFactory(ccf);
-		assertTrue(latch.await(10, TimeUnit.SECONDS));		
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		done.set(true);
 	}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java

@@ -32,7 +32,6 @@ import java.util.concurrent.TimeUnit;
 import javax.net.SocketFactory;
 
 import org.junit.Test;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
@@ -47,12 +46,12 @@ import org.springframework.integration.ip.util.SocketTestUtils;
 public class TcpNioConnectionReadTests {
 
 	private CountDownLatch latch = new CountDownLatch(1);
-	
+
 	private AbstractServerConnectionFactory getConnectionFactory(int port,
 			AbstractByteArraySerializer serializer, TcpListener listener) throws Exception {
 		return getConnectionFactory(port, serializer, listener, null);
 	}
-	
+
 	private AbstractServerConnectionFactory getConnectionFactory(int port,
 			AbstractByteArraySerializer serializer, TcpListener listener, TcpSender sender) throws Exception {
 		AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
@@ -72,7 +71,7 @@ public class TcpNioConnectionReadTests {
 		}
 		return scf;
 	}
-	
+
 	/**
 	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
 	 */
@@ -90,17 +89,17 @@ public class TcpNioConnectionReadTests {
 				return false;
 			}
 		});
-		
+
 		// Fire up the sender.
-		
+
 		SocketTestUtils.testSendLength(port, latch);
 		latch.countDown();
 		assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
 		assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
 		assertEquals("Did not receive data", 2, responses.size());
-		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, 
+		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
 						         new String(((Message<byte[]>) responses.get(0)).getPayload()));
-		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, 
+		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
 		         new String(((Message<byte[]>) responses.get(1)).getPayload()));
 		scf.close();
 	}
@@ -124,20 +123,20 @@ public class TcpNioConnectionReadTests {
 				return false;
 			}
 		});
-		
+
 		int howMany = 2;
-		scf.setPoolSize(howMany + 5);
+		scf.setBacklog(howMany + 5);
 		// Fire up the sender.
 		SocketTestUtils.testSendFragmented(port, howMany, false);
 		assertTrue(semaphore.tryAcquire(howMany, 20000, TimeUnit.MILLISECONDS));
 		assertEquals("Expected", howMany, responses.size());
 		for (int i = 0; i < howMany; i++) {
-			assertEquals("Data", "xx", 
+			assertEquals("Data", "xx",
 				new String(((Message<byte[]>) responses.get(0)).getPayload()));
 		}
 		scf.close();
 	}
-	
+
 	/**
 	 * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
 	 */
@@ -155,17 +154,17 @@ public class TcpNioConnectionReadTests {
 				return false;
 			}
 		});
-		
+
 		// Fire up the sender.
-		
+
 		SocketTestUtils.testSendStxEtx(port, latch);
 		latch.countDown();
 		assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
 		assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
 		assertEquals("Did not receive data", 2, responses.size());
-		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, 
+		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
 						         new String(((Message<byte[]>) responses.get(0)).getPayload()));
-		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, 
+		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
 		         new String(((Message<byte[]>) responses.get(1)).getPayload()));
 		scf.close();
 	}
@@ -187,17 +186,17 @@ public class TcpNioConnectionReadTests {
 				return false;
 			}
 		});
-		
+
 		// Fire up the sender.
-		
+
 		SocketTestUtils.testSendCrLf(port, latch);
 		latch.countDown();
 		assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
 		assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
 		assertEquals("Did not receive data", 2, responses.size());
-		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, 
+		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
 						         new String(((Message<byte[]>) responses.get(0)).getPayload()));
-		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, 
+		assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
 		         new String(((Message<byte[]>) responses.get(1)).getPayload()));
 		scf.close();
 	}
@@ -229,9 +228,9 @@ public class TcpNioConnectionReadTests {
 				semaphore.release();
 			}
 		});
-		
+
 		// Fire up the sender.
-		
+
 		SocketTestUtils.testSendLengthOverflow(port);
 		whileOpen(semaphore, added);
 		assertEquals(1, added.size());
@@ -268,9 +267,9 @@ public class TcpNioConnectionReadTests {
 				semaphore.release();
 			}
 		});
-		
+
 		// Fire up the sender.
-		
+
 		SocketTestUtils.testSendStxEtxOverflow(port);
 		whileOpen(semaphore, added);
 		assertEquals(1, added.size());
@@ -307,9 +306,9 @@ public class TcpNioConnectionReadTests {
 				semaphore.release();
 			}
 		});
-		
+
 		// Fire up the sender.
-		
+
 		SocketTestUtils.testSendCrLfOverflow(port);
 		whileOpen(semaphore, added);
 		assertEquals(1, added.size());
@@ -320,7 +319,7 @@ public class TcpNioConnectionReadTests {
 
 	/**
 	 * Tests socket closure when no data received.
-	 * 
+	 *
 	 * @throws Exception
 	 */
 	@Test
@@ -359,7 +358,7 @@ public class TcpNioConnectionReadTests {
 
 	/**
 	 * Tests socket closure when no data received.
-	 * 
+	 *
 	 * @throws Exception
 	 */
 	@Test
@@ -399,7 +398,7 @@ public class TcpNioConnectionReadTests {
 
 	/**
 	 * Tests socket closure when mid-message
-	 * 
+	 *
 	 * @throws Exception
 	 */
 	@Test
@@ -407,10 +406,10 @@ public class TcpNioConnectionReadTests {
 		ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
 		testClosureMidMessageGuts(serializer, "xx");
 	}
-	
+
 	/**
 	 * Tests socket closure when mid-message
-	 * 
+	 *
 	 * @throws Exception
 	 */
 
@@ -422,7 +421,7 @@ public class TcpNioConnectionReadTests {
 
 	/**
 	 * Tests socket closure when mid-message
-	 * 
+	 *
 	 * @throws Exception
 	 */
 

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java

@@ -19,15 +19,19 @@ package org.springframework.integration.ip.tcp.connection;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketTimeoutException;
+import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
@@ -36,19 +40,30 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import javax.net.ServerSocketFactory;
 
 import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.springframework.integration.Message;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
 import org.springframework.integration.ip.util.SocketTestUtils;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.ReflectionUtils.FieldCallback;
 import org.springframework.util.ReflectionUtils.FieldFilter;
+import org.springframework.util.ReflectionUtils.MethodCallback;
+import org.springframework.util.ReflectionUtils.MethodFilter;
 
 
 /**
@@ -84,7 +99,7 @@ public class TcpNioConnectionTests {
 			TcpConnection connection = factory.getConnection();
 			connection.send(MessageBuilder.withPayload(new byte[1000000]).build());
 		} catch (Exception e) {
-			assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() + 
+			assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() +
 					   ":" + e.getMessage(), e instanceof SocketTimeoutException);
 		}
 	}
@@ -102,7 +117,7 @@ public class TcpNioConnectionTests {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
 					latch.countDown();
 					Socket socket = server.accept();
-					byte[] b = new byte[6]; 
+					byte[] b = new byte[6];
 					readFully(socket.getInputStream(), b);
 					// block to cause timeout on read.
 					server.accept();
@@ -127,7 +142,7 @@ public class TcpNioConnectionTests {
 			fail("Unexpected exception " + e);
 		}
 	}
-	
+
 	@Test
 	public void testMemoryLeak() throws Exception {
 		final int port = SocketTestUtils.findAvailableServerSocket();
@@ -234,10 +249,108 @@ public class TcpNioConnectionTests {
 		assertEquals(0, TestUtils.getPropertyValue(factory, "connections", List.class).size());
 	}
 
+	@Test
+	public void testInsufficientThreads() throws Exception {
+		final ExecutorService exec = Executors.newFixedThreadPool(2);
+		Future<Object> future = exec.submit(new Callable<Object>() {
+			public Object call() throws Exception {
+				SocketChannel channel = mock(SocketChannel.class);
+				Socket socket = mock(Socket.class);
+				Mockito.when(channel.socket()).thenReturn(socket);
+				doAnswer(new Answer<Integer>() {
+					public Integer answer(InvocationOnMock invocation) throws Throwable {
+						ByteBuffer buffer = (ByteBuffer) invocation.getArguments()[0];
+						buffer.position(1025);
+						return 1025;
+					}
+				}).when(channel).read(Mockito.any(ByteBuffer.class));
+				final TcpNioConnection connection = new TcpNioConnection(channel, false, false);
+				connection.setTaskExecutor(exec);
+				connection.setPipeTimeout(200);
+				ReflectionUtils.doWithMethods(TcpNioConnection.class, new MethodCallback() {
+					public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+						method.setAccessible(true);
+						try {
+							method.invoke(connection, (Object[]) null);
+						}
+						catch (InvocationTargetException e) {
+							throw (RuntimeException) e.getCause();
+						}
+					}
+				}, new MethodFilter() {
+					public boolean matches(Method method) {
+						return method.getName().equals("doRead");
+					}
+				});
+				return null;
+			}
+		});
+		try {
+			Object o = future.get(10, TimeUnit.SECONDS);
+			fail("Expected exception, got " + o);
+		}
+		catch (ExecutionException e) {
+			assertEquals("Timed out writing to pipe, probably due to insufficient threads in " +
+					"a fixed thread pool; consider increasing this task executor pool size", e.getCause()
+					.getMessage());
+		}
+	}
+
+	@Test
+	public void testSufficientThreads() throws Exception {
+		final ExecutorService exec = Executors.newFixedThreadPool(3);
+		final CountDownLatch messageLatch = new CountDownLatch(1);
+		Future<Object> future = exec.submit(new Callable<Object>() {
+			public Object call() throws Exception {
+				SocketChannel channel = mock(SocketChannel.class);
+				Socket socket = mock(Socket.class);
+				Mockito.when(channel.socket()).thenReturn(socket);
+				doAnswer(new Answer<Integer>() {
+					public Integer answer(InvocationOnMock invocation) throws Throwable {
+						ByteBuffer buffer = (ByteBuffer) invocation.getArguments()[0];
+						buffer.position(1025);
+						buffer.put((byte) '\r');
+						buffer.put((byte) '\n');
+						return 1027;
+					}
+				}).when(channel).read(Mockito.any(ByteBuffer.class));
+				final TcpNioConnection connection = new TcpNioConnection(channel, false, false);
+				connection.setTaskExecutor(exec);
+				connection.registerListener(new TcpListener(){
+					public boolean onMessage(Message<?> message) {
+						System.out.println(message);
+						messageLatch.countDown();
+						return false;
+					}
+				});
+				connection.setMapper(new TcpMessageMapper());
+				connection.setDeserializer(new ByteArrayCrLfSerializer());
+				ReflectionUtils.doWithMethods(TcpNioConnection.class, new MethodCallback() {
+					public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
+						method.setAccessible(true);
+						try {
+							method.invoke(connection, (Object[]) null);
+						}
+						catch (InvocationTargetException e) {
+							throw (RuntimeException) e.getCause();
+						}
+					}
+				}, new MethodFilter() {
+					public boolean matches(Method method) {
+						return method.getName().equals("doRead");
+					}
+				});
+				return null;
+			}
+		});
+		future.get(60, TimeUnit.SECONDS);
+		assertTrue(messageLatch.await(10, TimeUnit.SECONDS));
+	}
+
 	private void readFully(InputStream is, byte[] buff) throws IOException {
 		for (int i = 0; i < buff.length; i++) {
 			buff[i] = (byte) is.read();
 		}
 	}
-	
+
 }

src/reference/docbook/ip.xml

@@ -798,30 +798,30 @@
    </para>
    <para><emphasis>Pool Size</emphasis></para>
    <para>
-    When using NIO, it is important to understand how threads are used, in order set the pool-size
-    appropriately. One thread from the pool is used to handle all socket events (e.g. data is
-    ready to be read). This thread is not available for other tasks. When data are ready to be
-    read, this thread dispatches the actual I/O to another thread from the pool, which reads
-    from the channel and writes the data to a temporary buffer; if this read is the start of
-    a new message, a third thread is used to read from that buffer to assemble the data into
-    a message.
+    The pool size attribute is no longer used; previously, it specified the size
+    of the default thread pool when a task-executor was not specified. It was also
+    used to set the connection backlog on server sockets. The first function is
+    no longer needed (see below); the second function is replaced by the
+    <emphasis>backlog</emphasis> attribute.
    </para>
    <para>
-    If there is not enough room in the temporary buffer to receive the newly read data, the
-    reader thread will block until the assembler thread consumes some data. If the pool is
-    exhausted, this will cause a deadlock, until another thread becomes available. The temporary
-    buffer is currently 1024 bytes. In the simplest case, with one connection, and data greater
-    than 1024 bytes, a pool-size of 2 will cause this deadlock to occur because a thread will
-    never be made available.
-   </para>
-   <para>
-    For this reason, when using NIO, the pool-size should be set to a minimum of 3. This does not
-    mean you have to reserve 2 threads for each socket because, aside from the selector thread,
-    the threads in the pool are shared across all the connections. The actual pool-size needed
-    will depend on a number of factors including the number of active connections, how
-    much utilization there is on those connections, and how long message processing takes when
-    a new message is received.
+    Previously, when using a fixed thread pool task executor (which was the default), with NIO, it
+    was possible to get a deadlock and processing would stop. The problem occurred when
+    a buffer was full, a thread reading from the socket was trying to add more data
+    to the buffer, and there were no threads available to make space in the buffer.
+    This only occurred with a very small pool size, but it could be possible under
+    extreme conditions. Since 2.2, two changes have eliminated this problem. First,
+    the default task executor is a cached thread pool executor. Second, deadlock
+    detection logic has been added such that if thread starvation occurs, instead of
+    deadlocking, an exception is thrown, thus releasing the deadlocked resources.
    </para>
+   <note>
+    Now that the default task executor is unbounded, it is possible that an out of
+    memory condition might occur with high rates of incoming messages, if message
+    processing takes extended time. If your application exhibits this type of
+    behavior, you are advised to use a pooled task executor with an appropriate
+    pool size.
+   </note>
   </section>
   <section id="ssl-tls">
     <title>SSL/TLS Support</title>
@@ -1136,9 +1136,8 @@
               <entry></entry>
               <entry>
                      Specifies a specific Executor to be used for socket handling. If not supplied, an internal
-                     pooled executor will be used. Needed on some platforms that require the use of specific
-                     task executors such as a WorkManagerTaskExecutor. See pool-size for thread
-                     requirements, depending on other options.</entry>
+                     cached thread executor will be used. Needed on some platforms that require the use of specific
+                     task executors such as a WorkManagerTaskExecutor.</entry>
             </row>
             <row>
               <entry>single-use</entry>
@@ -1150,16 +1149,20 @@
             </row>
             <row>
               <entry>pool-size</entry>
-              <entry>Y</entry>
+              <entry>N</entry>
+              <entry>N</entry>
+              <entry></entry>
+              <entry>This attribute is no longer used. For backward
+                     compatibility, it sets the backlog but users should
+                     use backlog to specify the
+                     connection backlog in server factories</entry>
+            </row>
+            <row>
+              <entry>backlog</entry>
+              <entry>N</entry>
               <entry>Y</entry>
               <entry></entry>
-              <entry>Specifies the concurrency.  For tcp, not using nio, specifies the
-                     number of concurrent connections supported by the adapter. For tcp,
-                     using nio, it should be set to a minimum of 3; see 'Pool Size' in
-                     <xref linkend="note_nio" />.
-                     It only applies in this sense if task-executor is not configured.
-                     However, pool-size is also used for the server socket backlog,
-                     regardless of whether an external task executor is used. Defaults to 5.</entry>
+              <entry>Sets the connection backlog for server factories.</entry>
             </row>
             <row>
               <entry>lookup-host</entry>