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(); - } - -}