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(); + } + }