diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index a446a757aa..0581fc75b5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -51,11 +51,11 @@ import org.springframework.util.Assert; */ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler implements TcpSender, TcpListener, SmartLifecycle { - private volatile AbstractConnectionFactory connectionFactory; + private volatile AbstractClientConnectionFactory connectionFactory; - private Map pendingReplies = new ConcurrentHashMap(); + private final Map pendingReplies = new ConcurrentHashMap(); - private Semaphore semaphore = new Semaphore(1, true); + private final Semaphore semaphore = new Semaphore(1, true); private volatile long remoteTimeout = 10000L; @@ -100,6 +100,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp Assert.notNull(connectionFactory, this.getClass().getName() + " requires a client connection factory"); boolean haveSemaphore = false; + String connectionId = null; try { boolean singleUseConnection = this.connectionFactory.isSingleUse(); if (!singleUseConnection) { @@ -114,13 +115,19 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp } TcpConnection connection = this.connectionFactory.getConnection(); AsyncReply reply = new AsyncReply(); - pendingReplies.put(connection.getConnectionId(), reply); + connectionId = connection.getConnectionId(); + pendingReplies.put(connectionId, reply); if (logger.isDebugEnabled()) { logger.debug("Added " + connection.getConnectionId()); } connection.send(requestMessage); Message replyMessage = reply.getReply(); if (replyMessage == null) { + if (logger.isDebugEnabled()) { + logger.debug("Remote Timeout on " + connection.getConnectionId()); + } + // The connection is dirty - force it closed. + this.connectionFactory.forceClose(connection); throw new MessageTimeoutException(requestMessage, "Timed out waiting for response"); } if (logger.isDebugEnabled()) { @@ -136,6 +143,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp throw new MessagingException("Failed to send or receive", e); } finally { + if (connectionId != null) { + pendingReplies.remove(connectionId); + } if (haveSemaphore) { this.semaphore.release(); if (logger.isDebugEnabled()) { @@ -161,9 +171,10 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp } public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { + // TODO: In 3.0 Change parameter type to AbstractClientConnectionFactory Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory, this.getClass().getName() + " requires a client connection factory"); - this.connectionFactory = connectionFactory; + this.connectionFactory = (AbstractClientConnectionFactory) connectionFactory; connectionFactory.registerListener(this); connectionFactory.registerSender(this); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 7208a375e9..244d66bf0c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -96,4 +96,16 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection return theConnection; } + /** + * Force close the connection and null the field if it's + * a shared connection. + * @param connection + */ + public void forceClose(TcpConnection connection) { + if (this.theConnection == connection) { + this.theConnection = null; + } + connection.close(); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index 1621abd205..c6a823b949 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -21,11 +21,15 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.net.ServerSocket; import java.net.Socket; +import java.util.ArrayList; import java.util.HashSet; +import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; @@ -34,9 +38,12 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; 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; @@ -55,6 +62,8 @@ import org.springframework.integration.test.util.TestUtils; */ public class TcpOutboundGatewayTests { + private final Log logger = LogFactory.getLog(this.getClass()); + @Test public void testGoodNetSingle() throws Exception { final int port = SocketUtils.findAvailableServerSocket(); @@ -223,8 +232,6 @@ public class TcpOutboundGatewayTests { for (int i = 0; i < 2; i++) { try { results[i].get(); - } catch (InterruptedException e) { - } catch (ExecutionException e) { if (timeouts > 0) { fail("Unexpected " + e.getMessage()); @@ -248,5 +255,116 @@ public class TcpOutboundGatewayTests { done.set(true); } + /** + * 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(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + /* + * The payload of the last message received by the remote side; + * used to verify the correct response is received. + */ + final AtomicReference lastReceived = new AtomicReference(); + final CountDownLatch serverLatch = new CountDownLatch(2); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + while (!done.get()) { + Socket socket = server.accept(); + i++; + while (!socket.isClosed()) { + try { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + String request = (String) ois.readObject(); + logger.debug("Read " + request); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + if (i < 2) { + Thread.sleep(1000); + } + oos.writeObject(request.replace("Test", "Reply")); + logger.debug("Replied to " + request); + lastReceived.set(request); + serverLatch.countDown(); + } + catch (IOException e) { + socket.close(); + } + } + } + } 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); + gateway.setRequestTimeout(Integer.MAX_VALUE); + QueueChannel replyChannel = new QueueChannel(); + gateway.setRequiresReply(true); + gateway.setOutputChannel(replyChannel); + gateway.setRemoteTimeout(500); + @SuppressWarnings("unchecked") + Future[] results = new Future[2]; + for (int i = 0; i < 2; i++) { + final int j = i; + results[j] = (Executors.newSingleThreadExecutor().submit(new Callable() { + public Integer call() throws Exception { + // increase the timeout after the first send + if (j > 0) { + gateway.setRemoteTimeout(5000); + } + gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build()); + return j; + } + })); + Thread.sleep(50); + } + // wait until the server side has processed both requests + assertTrue(serverLatch.await(10, TimeUnit.SECONDS)); + List replies = new ArrayList(); + int timeouts = 0; + for (int i = 0; i < 2; i++) { + try { + int result = results[i].get(); + String reply = (String) replyChannel.receive(1000).getPayload(); + logger.debug(i + " got " + result + " " + reply); + replies.add(reply); + } catch (ExecutionException e) { + if (timeouts >= 2) { + fail("Unexpected " + e.getMessage()); + } else { + assertNotNull(e.getCause()); + assertTrue(e.getCause() instanceof MessageTimeoutException); + } + timeouts++; + continue; + } + } + assertEquals("Expected exactly one ExecutionException", 1, timeouts); + assertEquals(1, replies.size()); + assertEquals(lastReceived.get().replace("Test", "Reply"), replies.get(0)); + done.set(true); + assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size()); + } }