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

Merge pull request #698 from garyrussell/INT-2860

Browse Source 
* INT-2860: Improve TCP Connection Timeout
This commit is contained in:
Mark Fisher
2013-01-22 14:29:41 -05:00
parent be29f909ba e4722de464
commit 3245fa58dc
7 changed files with 482 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

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

140
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -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<Object>) 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;
}
}

15
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -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<Object>) 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.
*

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

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

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

@@ -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

345
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java Normal file
View File

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

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

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