INT-3146 TCP SO Timeout with Caching Client CF

https://jira.springsource.org/browse/INT-3146

Incompatibility of using socket timeouts with a caching client
connection factory.

When a socket option timeout (soTimeout) is set on a TCP connection
and the timeout occurs, the socket is closed.

When a connection is intercepted, the close is performed through
the interceptor (for example to allow a closing handshake before the
physical close).

However, the caching client connection factory is implemented
using an interceptor, which returns the underlying connection to
the cache pool (for reuse). In the case of a TcpNetConnection,
the reader thread has terminated meaning that, the next time the connection
is used, no reply will ever be received.

The work-around (when using a gateway) is to use the 'remote-timeout'
attribute instead of relying on the soTimeout. There is no
work around when using collaborating channel adapters with a Net
connection. Using NIO works because there is no reader thread
in that case, but the socket is never closed on a timeout.

Always physically close the connection whenever an exception occurs
even if the close was delegated to an interceptor.

Add test cases for Net and NIO implementations.

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

@@ -143,7 +143,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 				if (logger.isDebugEnabled()) {
 					logger.debug("Closing single-use connection" + this.getConnectionId());
 				}
-				this.closeConnection();
+				this.closeConnection(false);
 			}
 		}
 	}
@@ -165,7 +165,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	 * If we have been intercepted, propagate the close from the outermost interceptor;
 	 * otherwise, just call close().
 	 */
-	protected void closeConnection() {
+	protected void closeConnection(boolean isException) {
 		if (!(this.listener instanceof TcpConnectionInterceptor)) {
 			close();
 			return;
@@ -175,6 +175,10 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 			outerInterceptor = (TcpConnectionInterceptor) outerInterceptor.getListener();
 		}
 		outerInterceptor.close();
+		if (isException) {
+			// ensure physical close in case the interceptor did not close
+			this.close();
+		}
 	}
 
 	/**

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

@@ -170,7 +170,7 @@ public class TcpNetConnection extends TcpConnectionSupport {
 				catch (NoListenerException nle) {
 					if (singleUse) {
 						logger.debug("Closing single use socket after inbound message " + this.getConnectionId());
-						this.closeConnection();
+						this.closeConnection(true);
 						okToRun = false;
 					} else {
 						logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
@@ -186,7 +186,7 @@ public class TcpNetConnection extends TcpConnectionSupport {
 				 */
 				if (singleUse && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) {
 					logger.debug("Closing single use socket after inbound message " + this.getConnectionId());
-					this.closeConnection();
+					this.closeConnection(false);
 					okToRun = false;
 				}
 			}
@@ -212,12 +212,11 @@ public class TcpNetConnection extends TcpConnectionSupport {
 			}
 			catch (SocketException e1) {
 				logger.error("Error accessing soTimeout", e1);
-				doClose = true;
 			}
 		}
 		if (doClose) {
 			boolean noReadErrorOnClose = this.isNoReadErrorOnClose();
-			this.closeConnection();
+			this.closeConnection(true);
 			if (!(e instanceof SoftEndOfStreamException)) {
 				if (e instanceof SocketTimeoutException && this.isSingleUse()) {
 					if (logger.isDebugEnabled()) {

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

@@ -228,7 +228,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 								     ":" + e.getCause() + ":" + e.getMessage());
 					}
 				}
-				this.closeConnection();
+				this.closeConnection(true);
 				this.sendExceptionToListener(e);
 				return;
 			}
@@ -268,8 +268,9 @@ public class TcpNioConnection extends TcpConnectionSupport {
 		Message<?> message = null;
 		try {
 			message = this.getMapper().toMessage(this);
-		} catch (Exception e) {
-			this.closeConnection();
+		}
+		catch (Exception e) {
+			this.closeConnection(true);
 			if (e instanceof SocketTimeoutException && this.isSingleUse()) {
 				if (logger.isDebugEnabled()) {
 					logger.debug("Closing single use socket after timeout " + this.getConnectionId());
@@ -290,13 +291,14 @@ public class TcpNioConnection extends TcpConnectionSupport {
 			if (message != null) {
 				intercepted = getListener().onMessage(message);
 			}
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			if (e instanceof NoListenerException) {
 				if (this.isSingleUse()) {
 					if (logger.isDebugEnabled()) {
 						logger.debug("Closing single use channel after inbound message " + this.getConnectionId());
 					}
-					this.closeConnection();
+					this.closeConnection(true);
 				}
 			}
 			else {
@@ -312,7 +314,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Closing single use channel after inbound message " + this.getConnectionId());
 			}
-			this.closeConnection();
+			this.closeConnection(false);
 		}
 	}
 
@@ -334,7 +336,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 			int len = this.socketChannel.read(this.rawBuffer);
 			if (len < 0) {
 				this.writingToPipe = false;
-				this.closeConnection();
+				this.closeConnection(true);
 			}
 			if (logger.isTraceEnabled()) {
 				logger.trace("After read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
@@ -412,16 +414,18 @@ public class TcpNioConnection extends TcpConnectionSupport {
 		}
 		try {
 			doRead();
-		} catch (ClosedChannelException cce) {
+		}
+		catch (ClosedChannelException cce) {
 			if (logger.isDebugEnabled()) {
 				logger.debug(this.getConnectionId() + " Channel is closed");
 			}
-			this.closeConnection();
-		} catch (Exception e) {
+			this.closeConnection(true);
+		}
+		catch (Exception e) {
 			logger.error("Exception on Read " +
 					     this.getConnectionId() + " " +
 					     e.getMessage(), e);
-			this.closeConnection();
+			this.closeConnection(true);
 		}
 	}
 
@@ -429,7 +433,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 	 * Close the socket due to timeout.
 	 */
 	void timeout() {
-		this.closeConnection();
+		this.closeConnection(true);
 	}
 
 	/**

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml

@@ -7,12 +7,13 @@
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
 
+	<bean id="tcpIpUtils" class="org.springframework.integration.test.util.SocketUtils" />
 
 	<int-ip:tcp-connection-factory
 		id="scf"
 		type="server"
 		so-timeout="60000"
-		port="9876"/>
+		port="#{tcpIpUtils.findAvailableServerSocket()}"/>
 
 	<int-ip:tcp-inbound-channel-adapter
 		connection-factory="scf"
@@ -32,7 +33,7 @@
 		id="ccf"
 		type="client"
 		host="localhost"
-		port="9876"
+		port="#{scf.port}"
 		so-timeout="60000"
 
 	/>
@@ -53,7 +54,7 @@
 		id="gateway.ccf"
 		type="client"
 		host="localhost"
-		port="9876"
+		port="#{scf.port}"
 		so-timeout="60000"
 	/>
 

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

@@ -16,6 +16,7 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertSame;
@@ -353,6 +354,28 @@ public class CachingClientConnectionFactoryTests {
 		okToRun.set(false);
 	}
 
+	@Test
+	public void testCloseOnTimeoutNet() throws Exception {
+		TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost", serverCf.getPort());
+		testCloseOnTimeoutGuts(cf);
+	}
+
+	@Test
+	public void testCloseOnTimeoutNio() throws Exception {
+		TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost", serverCf.getPort());
+		testCloseOnTimeoutGuts(cf);
+	}
+
+	private void testCloseOnTimeoutGuts(AbstractClientConnectionFactory cf) throws Exception, InterruptedException {
+		cf.setSoTimeout(100);
+		CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(cf, 1);
+		cccf.start();
+		TcpConnection connection = cccf.getConnection();
+		Thread.sleep(200);
+		assertFalse(connection.isOpen());
+		cccf.stop();
+	}
+
 	@Test
 	public void testCachedFailover() throws Exception {
 		// Failover