Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3111 Fix Close Cached TCP Connection

Browse Source 
A TCP Connection is closed when an outbound gateway times out.

However, with the CachingClientConnectionFactory, the connection
was returned to the pool instead of being physically closed.

Override forceClose() and physically close the connection so the
next time it is retrieved from the pool it is detected as stale
and re-established.

Refactor test case and invoke it twice; once with a regular connection
factory and once with a cached factory.

Remove an unused logger.
This commit is contained in:
Gary Russell
2013-08-18 10:29:18 -04:00
parent 56af118656
commit b8980d4064
2 changed files with 51 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

@@ -17,8 +17,6 @@ package org.springframework.integration.ip.tcp.connection;
import java.util.concurrent.Executor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.Message;
@@ -34,8 +32,6 @@ import org.springframework.integration.util.SimplePool;
*/
public class CachingClientConnectionFactory extends AbstractClientConnectionFactory {
private final Log logger = LogFactory.getLog(this.getClass());
private final AbstractClientConnectionFactory targetConnectionFactory;
private final SimplePool<TcpConnectionSupport> pool;
@@ -160,6 +156,10 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
return true; // true so the single-use connection doesn't close itself
}
private void physicallyClose() {
this.getTheConnection().close();
}
}
///////////////// DELEGATE METHODS ///////////////////////
@@ -357,6 +357,16 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
return targetConnectionFactory.isLookupHost();
}
@Override
public void forceClose(TcpConnection connection) {
if (connection instanceof CachedConnection) {
((CachedConnection) connection).physicallyClose();
}
// will be returned to pool but stale, so will be re-established
super.forceClose(connection);
}
@Override
public void start() {
this.setActive(true);

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

@@ -45,12 +45,15 @@ import javax.net.ServerSocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
@@ -255,15 +258,39 @@ public class TcpOutboundGatewayTests {
done.set(true);
}
@Test
public void testGoodNetGWTimeout() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractClientConnectionFactory ccf = buildCF(port);
ccf.start();
testGoodNetGWTimeoutGuts(port, ccf);
}
@Test
public void testGoodNetGWTimeoutCached() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractClientConnectionFactory ccf = buildCF(port);
CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(ccf, 1);
cccf.start();
testGoodNetGWTimeoutGuts(port, cccf);
}
private AbstractClientConnectionFactory buildCF(final int port) {
AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.setSingleUse(false);
return ccf;
}
/**
* Sends 2 concurrent messages on a shared connection. The GW single threads
* these requests. The first will timeout; the second should receive its
* own response, not that for the first.
* @throws Exception
*/
@Test
public void testGoodNetGWTimeout() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
private void testGoodNetGWTimeoutGuts(final int port, AbstractConnectionFactory ccf) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
/*
@@ -298,23 +325,19 @@ public class TcpOutboundGatewayTests {
serverLatch.countDown();
}
catch (IOException e) {
logger.debug("error on write " + e.getClass().getSimpleName());
socket.close();
}
}
}
} catch (Exception e) {
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.setSingleUse(false);
ccf.start();
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
final TcpOutboundGateway gateway = new TcpOutboundGateway();
gateway.setConnectionFactory(ccf);
@@ -349,10 +372,12 @@ public class TcpOutboundGatewayTests {
String reply = (String) replyChannel.receive(1000).getPayload();
logger.debug(i + " got " + result + " " + reply);
replies.add(reply);
} catch (ExecutionException e) {
}
catch (ExecutionException e) {
if (timeouts >= 2) {
fail("Unexpected " + e.getMessage());
} else {
}
else {
assertNotNull(e.getCause());
assertTrue(e.getCause() instanceof MessageTimeoutException);
}