INT-2860 Improve TCP Connection Timeout

A connection used by a gateway may timeout prematurely.

Previously, the socket read timeout simply controlled when
a socket would be closed after the timeout occurred.

Consider a connection with so-timeout set to 10 seconds; the
application initializes at T+0 and the first message is sent,
with the server responding immediately; the timeout clock starts.
Next, a message is sent at T+5 to a service that takes 6 seconds
to respond. The connection will timeout at T+10 before the
response is received; the socket is closed and client does not
receive the response.

The solution is to wait 2 timeout cycles *IF* a message has
been sent within the current timeout.

Maintain a timer for the last send() operation. When a socket
timeout occurs, examine the last sent time; if within the
timeout, defer the close until the next timeout.

We cannot simply rely on the last send time because, when
using collaborating adapters, continuous sends (with no replies)
would defer the close indefinitely. Hence, the second test looking
to see if we have not had a successful read for the last 2 timeouts.

NIO does not directly support socket timeouts (because there is no
thread hanging on the read); instead, the timeout logic is
performed on the selector thread.

Rename DefaultTimeoutTests to ConnectionTimeoutTests.

Add tests (for both Socket and NIO connections) to assert the correct
operation when a send is performed within a timeout, as well as
when the server takes > 2x the timeout to respond.

INT-2860 Fix Typo in Exception Message

Error sending meeeage.

removed invalid comment from test (while merging)

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java

@@ -167,6 +167,7 @@ public class TcpInboundGatewayTests {
 		latch2.countDown();
 		assertTrue(latch3.await(10, TimeUnit.SECONDS));
 		assertTrue(done.get());
+		gateway.stop();
 	}
 
 	@Test

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java

@@ -131,6 +131,7 @@ public class TcpSendingMessageHandlerTests {
 		assertNotNull(mOut);
 		assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -244,6 +245,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(results.remove("Reply1"));
 		assertTrue(results.remove("Reply2"));
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -293,6 +295,7 @@ public class TcpSendingMessageHandlerTests {
 		assertNotNull(mOut);
 		assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -345,6 +348,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(results.remove("Reply1"));
 		assertTrue(results.remove("Reply2"));
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -397,6 +401,7 @@ public class TcpSendingMessageHandlerTests {
 		assertNotNull(mOut);
 		assertEquals("Reply2", new String((byte[]) mOut.getPayload()));
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -452,6 +457,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(results.remove("Reply1"));
 		assertTrue(results.remove("Reply2"));
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -500,6 +506,7 @@ public class TcpSendingMessageHandlerTests {
 		assertNotNull(mOut);
 		assertEquals("Reply2", mOut.getPayload());
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -551,6 +558,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(results.remove("Reply1"));
 		assertTrue(results.remove("Reply2"));
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -888,6 +896,7 @@ public class TcpSendingMessageHandlerTests {
 		assertNotNull(mOut);
 		assertEquals("Reply2", mOut.getPayload());
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -953,6 +962,7 @@ public class TcpSendingMessageHandlerTests {
 			assertTrue("Missing Reply" + i, results.remove("Reply" + i));
 		}
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -1010,6 +1020,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test
@@ -1066,6 +1077,7 @@ public class TcpSendingMessageHandlerTests {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		handler.handleMessage(MessageBuilder.withPayload("Test").build());
 		done.set(true);
+		ccf.stop();
 	}
 
 	@Test

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java

@@ -0,0 +1,345 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.net.Socket;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Test;
+import org.springframework.integration.Message;
+import org.springframework.integration.ip.util.TestingUtilities;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.SocketUtils;
+import org.springframework.integration.test.util.TestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class ConnectionTimeoutTests {
+
+	@Test
+	public void testDefaultTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				return false;
+			}
+		});
+		TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
+		client.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				return false;
+			}
+		});
+		server.start();
+		TestingUtilities.waitListening(server, null);
+		client.start();
+		TcpConnection connection = client.getConnection();
+		Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
+		// should default to 0 (infinite) timeout
+		assertEquals(0, socket.getSoTimeout());
+		connection.close();
+		server.stop();
+		client.stop();
+	}
+
+	@Test
+	public void testNetSimpleTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				return false;
+			}
+		});
+		TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
+		client.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				return false;
+			}
+		});
+		client.setSoTimeout(1000);
+		server.start();
+		TestingUtilities.waitListening(server, null);
+		client.start();
+		TcpConnection connection = client.getConnection();
+		Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
+		assertEquals(1000, socket.getSoTimeout());
+		Thread.sleep(1100);
+		assertFalse(connection.isOpen());
+		server.stop();
+		client.stop();
+	}
+
+	/**
+	 * Ensure we don't timeout on the read side (client) if we sent a message within the
+	 * current timeout.
+	 * @throws Exception
+	 */
+	@Test
+	public void testNetReplyNotTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				try {
+					Thread.sleep(1200);
+					serverConnection.get().send(message);
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+				}
+				return false;
+			}
+		});
+		server.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+				serverConnection.set(connection);
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
+		client.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				reply.set(message);
+				return false;
+			}
+		});
+		client.setSoTimeout(2000);
+		server.start();
+		TestingUtilities.waitListening(server, null);
+		client.start();
+		TcpConnection connection = client.getConnection();
+		Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
+		assertEquals(2000, socket.getSoTimeout());
+		Thread.sleep(1000);
+		connection.send(MessageBuilder.withPayload("foo").build());
+		Thread.sleep(1400);
+		assertNotNull(reply.get());
+		Thread.sleep(2200);
+		assertFalse(connection.isOpen());
+		server.stop();
+		client.stop();
+	}
+
+	/**
+	 * Ensure we don't timeout on the read side (client) if we sent a message within the
+	 * current timeout.
+	 * @throws Exception
+	 */
+	@Test
+	public void testNioReplyNotTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				try {
+					Thread.sleep(1200);
+					serverConnection.get().send(message);
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+				}
+				return false;
+			}
+		});
+		server.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+				serverConnection.set(connection);
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port);
+		client.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				reply.set(message);
+				return false;
+			}
+		});
+		client.setSoTimeout(2000);
+		server.start();
+		TestingUtilities.waitListening(server, null);
+		client.start();
+		TcpConnection connection = client.getConnection();
+		Thread.sleep(1000);
+		connection.send(MessageBuilder.withPayload("foo").build());
+		Thread.sleep(1400);
+		assertNotNull(reply.get());
+		Thread.sleep(4200);
+		assertFalse(connection.isOpen());
+		server.stop();
+		client.stop();
+	}
+
+	/**
+	 * Ensure we do timeout on the read side (client) if we sent a message within the
+	 * first timeout but the reply takes > 2 timeouts.
+	 * @throws Exception
+	 */
+	@Test
+	public void testNetReplyTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				try {
+					Thread.sleep(4200);
+					serverConnection.get().send(message);
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+				}
+				return false;
+			}
+		});
+		server.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+				serverConnection.set(connection);
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
+		client.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				reply.set(message);
+				return false;
+			}
+		});
+		client.setSoTimeout(2000);
+		server.start();
+		TestingUtilities.waitListening(server, null);
+		client.start();
+		TcpConnection connection = client.getConnection();
+		Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
+		assertEquals(2000, socket.getSoTimeout());
+		Thread.sleep(1000);
+		connection.send(MessageBuilder.withPayload("foo").build());
+		Thread.sleep(1400);
+		assertTrue(connection.isOpen());
+		Thread.sleep(2000);
+		assertNull(reply.get());
+		assertFalse(connection.isOpen());
+		server.stop();
+		client.stop();
+	}
+
+	/**
+	 * Ensure we do timeout on the read side (client) if we sent a message within the
+	 * first timeout but the reply takes > 2 timeouts.
+	 * @throws Exception
+	 */
+	@Test
+	public void testNioReplyTimeout() throws Exception {
+		int port = SocketUtils.findAvailableServerSocket();
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
+		final AtomicReference<TcpConnection> serverConnection = new AtomicReference<TcpConnection>();
+		server.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				try {
+					Thread.sleep(2100);
+					serverConnection.get().send(message);
+				}
+				catch (Exception e) {
+					e.printStackTrace();
+				}
+				return false;
+			}
+		});
+		server.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+				serverConnection.set(connection);
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port);
+		client.registerSender(new TcpSender() {
+			public void addNewConnection(TcpConnection connection) {
+			}
+			public void removeDeadConnection(TcpConnection connection) {
+			}
+		});
+		final AtomicReference<Message<?>> reply = new AtomicReference<Message<?>>();
+		client.registerListener(new TcpListener() {
+			public boolean onMessage(Message<?> message) {
+				reply.set(message);
+				return false;
+			}
+		});
+		client.setSoTimeout(1000);
+		server.start();
+		TestingUtilities.waitListening(server, null);
+		client.start();
+		TcpConnection connection = client.getConnection();
+		Thread.sleep(500);
+		connection.send(MessageBuilder.withPayload("foo").build());
+		Thread.sleep(700);
+		assertTrue(connection.isOpen());
+		Thread.sleep(1000);
+		assertNull(reply.get());
+		assertFalse(connection.isOpen());
+		server.stop();
+		client.stop();
+	}
+}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/DefaultTimeoutTests.java

@@ -1,68 +0,0 @@
-/*
- * Copyright 2002-2012 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.integration.ip.tcp.connection;
-
-import static org.junit.Assert.assertEquals;
-
-import java.net.Socket;
-
-import org.junit.Test;
-import org.springframework.integration.Message;
-import org.springframework.integration.ip.util.TestingUtilities;
-import org.springframework.integration.test.util.SocketUtils;
-import org.springframework.integration.test.util.TestUtils;
-
-/**
- * @author Gary Russell
- * @since 2.2
- *
- */
-public class DefaultTimeoutTests {
-
-	@Test
-	public void test() throws Exception {
-		int port = SocketUtils.findAvailableServerSocket();
-		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port);
-		server.registerListener(new TcpListener() {
-			public boolean onMessage(Message<?> message) {
-				return false;
-			}
-		});
-		TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port);
-		client.registerSender(new TcpSender() {
-			public void addNewConnection(TcpConnection connection) {
-			}
-			public void removeDeadConnection(TcpConnection connection) {
-			}
-		});
-		client.registerListener(new TcpListener() {
-			public boolean onMessage(Message<?> message) {
-				return false;
-			}
-		});
-		server.start();
-		TestingUtilities.waitListening(server, null);
-		client.start();
-		TcpConnection connection = client.getConnection();
-		Socket socket = TestUtils.getPropertyValue(connection, "socket", Socket.class);
-		// should default to 0 (infinite) timeout
-		assertEquals(0, socket.getSoTimeout());
-		connection.close();
-		server.stop();
-		client.stop();
-	}
-
-}