diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
index ab9a91b8db..01d2ed28b0 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
+++ b/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();
+ }
}
/**
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 204494b300..bbd71c52a0 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
@@ -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()) {
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 139f6ba061..25c0ff1e34 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
@@ -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);
}
/**
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 01673f9b04..ef2ad7e7ea 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;
@@ -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