From e4722de464acb1e83965ebc294fc2fe49cff4acb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 3 Jan 2013 10:17:10 -0500 Subject: [PATCH] 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) --- .../connection/AbstractConnectionFactory.java | 27 +- .../ip/tcp/connection/TcpNetConnection.java | 140 ++++--- .../ip/tcp/connection/TcpNioConnection.java | 15 +- .../ip/tcp/TcpInboundGatewayTests.java | 1 + .../ip/tcp/TcpSendingMessageHandlerTests.java | 12 + .../connection/ConnectionTimeoutTests.java | 345 ++++++++++++++++++ .../tcp/connection/DefaultTimeoutTests.java | 68 ---- 7 files changed, 482 insertions(+), 126 deletions(-) create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java delete mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/DefaultTimeoutTests.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index a6d7c9f109..62fa264585 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -536,11 +536,28 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport } else if (soTimeout > 0) { TcpNioConnection connection = connections.get(channel); - if (now - connection.getLastRead() > this.soTimeout) { - logger.warn("Timing out TcpNioConnection " + - this.port + " : " + - connection.getConnectionId()); - connection.timeout(); + if (now - connection.getLastRead() >= this.soTimeout) { + /* + * For client connections, we have to wait for 2 timeouts if the last + * send was within the current timeout. + */ + if (!connection.isServer() && + now - connection.getLastSend() < this.soTimeout && + now - connection.getLastRead() < this.soTimeout * 2) + { + if (logger.isDebugEnabled()) { + logger.debug("Skipping a connection timeout because we have a recent send " + + connection.getConnectionId()); + } + } + else { + if (logger.isWarnEnabled()) { + logger.warn("Timing out TcpNioConnection " + + this.port + " : " + + connection.getConnectionId()); + } + connection.timeout(); + } } } } 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 41f595c362..9e64f88a43 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 @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp.connection; import java.net.Socket; +import java.net.SocketException; import java.net.SocketTimeoutException; import org.springframework.core.serializer.Deserializer; @@ -37,6 +38,10 @@ public class TcpNetConnection extends AbstractTcpConnection { private boolean noReadErrorOnClose; + private volatile long lastRead = System.currentTimeMillis(); + + private volatile long lastSend; + /** * Constructs a TcpNetConnection for the socket. * @param socket the socket @@ -67,6 +72,7 @@ public class TcpNetConnection extends AbstractTcpConnection { @SuppressWarnings("unchecked") public synchronized void send(Message message) throws Exception { Object object = this.getMapper().fromMessage(message); + this.lastSend = System.currentTimeMillis(); ((Serializer) this.getSerializer()).serialize(object, this.socket.getOutputStream()); this.afterSend(message); } @@ -96,54 +102,31 @@ public class TcpNetConnection extends AbstractTcpConnection { logger.debug("TcpListener exiting - no listener and not single use"); return; } - Message message = null; boolean okToRun = true; logger.debug("Reading..."); boolean intercepted = false; while (okToRun) { + Message message = null; try { message = this.getMapper().toMessage(this); + this.lastRead = System.currentTimeMillis(); } catch (Exception e) { - this.closeConnection(); - if (!(e instanceof SoftEndOfStreamException)) { - if (e instanceof SocketTimeoutException && singleUse) { - logger.debug("Closing single use socket after timeout"); - } else { - if (this.noReadErrorOnClose) { - if (logger.isTraceEnabled()) { - logger.trace("Read exception " + - this.getConnectionId(), e); - } - else if (logger.isDebugEnabled()) { - logger.debug("Read exception " + - this.getConnectionId() + " " + - e.getClass().getSimpleName() + - ":" + e.getCause() + ":" + e.getMessage()); - } - } else if (logger.isTraceEnabled()) { - logger.error("Read exception " + - this.getConnectionId(), e); - } else { - logger.error("Read exception " + - this.getConnectionId() + " " + - e.getClass().getSimpleName() + - ":" + e.getCause() + ":" + e.getMessage()); - } + if (handleReadException(e)) { + okToRun = false; + } + } + if (okToRun && message != null) { + if (logger.isDebugEnabled()) { + logger.debug("Message received " + message); + } + try { + if (listener == null) { + logger.warn("Unexpected message - no inbound adapter registered with connection " + message); + continue; } + intercepted = this.getListener().onMessage(message); } - break; - } - if (logger.isDebugEnabled()) { - logger.debug("Message received " + message); - } - try { - if (listener == null) { - logger.warn("Unexpected message - no inbound adapter registered with connection " + message); - continue; - } - intercepted = this.getListener().onMessage(message); - } catch (Exception e) { - if (e instanceof NoListenerException) { + catch (NoListenerException nle) { if (singleUse) { logger.debug("Closing single use socket after inbound message " + this.getConnectionId()); this.closeConnection(); @@ -151,21 +134,76 @@ public class TcpNetConnection extends AbstractTcpConnection { } else { logger.warn("Unexpected message - no inbound adapter registered with connection " + message); } - } else { - logger.error("Exception sending meeeage: " + message, e); } - } - /* - * For single use sockets, we close after receipt if we are on the client - * side, and the data was not intercepted, - * or the server side has no outbound adapter registered - */ - if (singleUse && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) { - logger.debug("Closing single use socket after inbound message " + this.getConnectionId()); - this.closeConnection(); - okToRun = false; + catch (Exception e2) { + logger.error("Exception sending message: " + message, e2); + } + /* + * For single use sockets, we close after receipt if we are on the client + * side, and the data was not intercepted, + * or the server side has no outbound adapter registered + */ + if (singleUse && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) { + logger.debug("Closing single use socket after inbound message " + this.getConnectionId()); + this.closeConnection(); + okToRun = false; + } } } } + protected boolean handleReadException(Exception e) { + boolean doClose = true; + /* + * For client connections, we have to wait for 2 timeouts if the last + * send was within the current timeout. + */ + if (!this.isServer() && e instanceof SocketTimeoutException) { + long now = System.currentTimeMillis(); + try { + int soTimeout = this.socket.getSoTimeout(); + if (now - this.lastSend < soTimeout && now - this.lastRead < soTimeout * 2) { + doClose = false; + } + if (!doClose && logger.isDebugEnabled()) { + logger.debug("Skipping a socket timeout because we have a recent send " + this.getConnectionId()); + } + } + catch (SocketException e1) { + logger.error("Error accessing soTimeout", e1); + doClose = true; + } + } + if (doClose) { + this.closeConnection(); + if (!(e instanceof SoftEndOfStreamException)) { + if (e instanceof SocketTimeoutException && this.isSingleUse()) { + logger.debug("Closing single use socket after timeout"); + } else { + if (this.noReadErrorOnClose) { + if (logger.isTraceEnabled()) { + logger.trace("Read exception " + + this.getConnectionId(), e); + } + else if (logger.isDebugEnabled()) { + logger.debug("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + } + } else if (logger.isTraceEnabled()) { + logger.error("Read exception " + + this.getConnectionId(), e); + } else { + logger.error("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + } + } + } + } + return doClose; + } + } 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 b199707cdf..d80e080fb5 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 @@ -67,6 +67,8 @@ public class TcpNioConnection extends AbstractTcpConnection { private volatile long lastRead; + private volatile long lastSend; + private AtomicInteger executionControl = new AtomicInteger(); private volatile boolean writingToPipe; @@ -120,6 +122,7 @@ public class TcpNioConnection extends AbstractTcpConnection { public void send(Message message) throws Exception { synchronized(this.getMapper()) { Object object = this.getMapper().fromMessage(message); + this.lastSend = System.currentTimeMillis(); ((Serializer) this.getSerializer()).serialize(object, this.getChannelOutputStream()); this.afterSend(message); } @@ -255,8 +258,9 @@ public class TcpNioConnection extends AbstractTcpConnection { } this.closeConnection(); } - } else { - logger.error("Exception sending meeeage: " + message, e); + } + else { + logger.error("Exception sending message: " + message, e); } } /* @@ -421,6 +425,13 @@ public class TcpNioConnection extends AbstractTcpConnection { this.lastRead = lastRead; } + /** + * @return the time of the last send + */ + public long getLastSend() { + return lastSend; + } + /** * OutputStream to wrap a SocketChannel; implements timeout on write. * diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index af0fe84d69..f4940df4bc 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 78a039220b..db19e88043 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java new file mode 100644 index 0000000000..c379bcc474 --- /dev/null +++ b/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 serverConnection = new AtomicReference(); + 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> reply = new AtomicReference>(); + 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 serverConnection = new AtomicReference(); + 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> reply = new AtomicReference>(); + 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 serverConnection = new AtomicReference(); + 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> reply = new AtomicReference>(); + 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 serverConnection = new AtomicReference(); + 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> reply = new AtomicReference>(); + 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(); + } +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/DefaultTimeoutTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/DefaultTimeoutTests.java deleted file mode 100644 index b5f64fe0e5..0000000000 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/DefaultTimeoutTests.java +++ /dev/null @@ -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(); - } - -}