From 33ecb6a55c83781fe886281b5fada6014dbe6f24 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 24 Sep 2013 10:40:02 -0400 Subject: [PATCH] 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. Conflicts: spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java Resolved. INT-3146 Fix Test Timing Issue Wait for server connection factory to start listening for new tests. INT-3146 Fix Another Test Timing Issue Increase test timer for soTimeout detection. --- .../tcp/connection/AbstractTcpConnection.java | 8 ++++-- .../ip/tcp/connection/TcpNetConnection.java | 7 ++--- .../ip/tcp/connection/TcpNioConnection.java | 28 +++++++++++-------- ...ngClientConnectionFactoryTests-context.xml | 7 +++-- .../CachingClientConnectionFactoryTests.java | 28 +++++++++++++++++++ 5 files changed, 57 insertions(+), 21 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java index 5922451f28..6cdb72efd5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -105,7 +105,7 @@ public abstract class AbstractTcpConnection implements TcpConnection { if (logger.isDebugEnabled()) { logger.debug("Closing single-use connection" + this.getConnectionId()); } - this.closeConnection(); + this.closeConnection(false); } } } @@ -123,7 +123,7 @@ public abstract class AbstractTcpConnection 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; @@ -133,6 +133,10 @@ public abstract class AbstractTcpConnection implements TcpConnection { outerInterceptor = (TcpConnectionInterceptor) outerInterceptor.getListener(); } outerInterceptor.close(); + if (isException) { + // ensure physical close in case the interceptor did not close + this.close(); + } } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index 81ff41b9ea..8a5fa0c041 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -129,7 +129,7 @@ public class TcpNetConnection extends AbstractTcpConnection { 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); @@ -145,7 +145,7 @@ public class TcpNetConnection extends AbstractTcpConnection { */ 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; } } @@ -171,12 +171,11 @@ public class TcpNetConnection extends AbstractTcpConnection { } 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()) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index bb74923af6..2437589720 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -200,7 +200,7 @@ public class TcpNioConnection extends AbstractTcpConnection { ":" + e.getCause() + ":" + e.getMessage()); } } - this.closeConnection(); + this.closeConnection(true); return; } } finally { @@ -237,8 +237,9 @@ public class TcpNioConnection extends AbstractTcpConnection { 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()); @@ -259,13 +260,14 @@ public class TcpNioConnection extends AbstractTcpConnection { 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 { @@ -281,7 +283,7 @@ public class TcpNioConnection extends AbstractTcpConnection { if (logger.isDebugEnabled()) { logger.debug("Closing single use cbannel after inbound message " + this.getConnectionId()); } - this.closeConnection(); + this.closeConnection(false); } } @@ -303,7 +305,7 @@ public class TcpNioConnection extends AbstractTcpConnection { 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()); @@ -373,16 +375,18 @@ public class TcpNioConnection extends AbstractTcpConnection { } 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); } } @@ -390,7 +394,7 @@ public class TcpNioConnection extends AbstractTcpConnection { * Close the socket due to timeout. */ void timeout() { - this.closeConnection(); + this.closeConnection(true); } /** diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml index ce3986f640..522674071d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml +++ b/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"> + + port="#{tcpIpUtils.findAvailableServerSocket()}"/> @@ -53,7 +54,7 @@ id="gateway.ccf" type="client" host="localhost" - port="9876" + port="#{scf.port}" so-timeout="60000" /> diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index d490843fda..28edb5ef00 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/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; @@ -347,4 +348,31 @@ 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 { + TestingUtilities.waitListening(serverCf, null); + cf.setSoTimeout(100); + CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(cf, 1); + cccf.start(); + TcpConnection connection = cccf.getConnection(); + int n = 0; + while (n++ < 100 && connection.isOpen()) { + Thread.sleep(100); + } + assertFalse(connection.isOpen()); + cccf.stop(); + } + }