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

INT-3837: TCP GW - Propagate Socket Timeout

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3837

INT-3103 introduced exception propagation to waiting gateway threads.

However, `SocketTimeoutException`s were not propagated (in all cases
since 4.2 and for single-use sockets since 3.0).
This commit is contained in:
Gary Russell
2015-09-28 10:11:23 -04:00
parent c23aa70ec8
commit 4ab2c158b3
3 changed files with 122 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -595,8 +595,10 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
logger.warn("Timing out TcpNioConnection " +
connection.getConnectionId());
}
connection.publishConnectionExceptionEvent(new SocketTimeoutException("Timing out connection"));
SocketTimeoutException exception = new SocketTimeoutException("Timing out connection");
connection.publishConnectionExceptionEvent(exception);
connection.timeout();
connection.sendExceptionToListener(exception);
}
}
}

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

@@ -249,8 +249,8 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
}
this.sendExceptionToListener(e);
}
this.sendExceptionToListener(e);
}
}
return doClose;

120
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.ip.tcp;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
@@ -33,6 +34,7 @@ import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -53,7 +55,6 @@ import javax.net.ServerSocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.log4j.Level;
import org.hamcrest.Matchers;
import org.junit.Rule;
import org.junit.Test;
import org.mockito.Mockito;
@@ -662,11 +663,13 @@ public class TcpOutboundGatewayTests {
@Override
public void run() {
List<Socket> sockets = new ArrayList<Socket>();
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
while (!done.get()) {
Socket socket = server.accept();
sockets.add(socket);
while (!socket.isClosed()) {
try {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
@@ -687,6 +690,12 @@ public class TcpOutboundGatewayTests {
e.printStackTrace();
}
}
for (Socket socket : sockets) {
try {
socket.close();
}
catch (IOException e) {}
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
@@ -705,13 +714,120 @@ public class TcpOutboundGatewayTests {
fail("expected failure");
}
catch (Exception e) {
assertThat(e.getCause(), Matchers.instanceOf(EOFException.class));
assertThat(e.getCause(), instanceOf(EOFException.class));
}
assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size());
Message<?> reply = replyChannel.receive(0);
assertNull(reply);
done.set(true);
ccf.getConnection();
gateway.stop();
ccf.stop();
}
@Test
public void testNetGWPropagatesSocketTimeout() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(100);
ccf.setSingleUse(false);
ccf.start();
testGWPropagatesSocketTimeoutGuts(port, ccf);
}
@Test
public void testNioGWPropagatesSocketTimeout() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(100);
ccf.setSingleUse(false);
ccf.start();
testGWPropagatesSocketTimeoutGuts(port, ccf);
}
@Test
public void testNetGWPropagatesSocketTimeoutSingleUse() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(100);
ccf.setSingleUse(true);
ccf.start();
testGWPropagatesSocketTimeoutGuts(port, ccf);
}
@Test
public void testNioGWPropagatesSocketTimeoutSingleUse() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(100);
ccf.setSingleUse(true);
ccf.start();
testGWPropagatesSocketTimeoutGuts(port, ccf);
}
private void testGWPropagatesSocketTimeoutGuts(final int port, AbstractClientConnectionFactory ccf)
throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
List<Socket> sockets = new ArrayList<Socket>();
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
while (!done.get()) {
sockets.add(server.accept());
}
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
for (Socket socket : sockets) {
try {
socket.close();
}
catch (IOException e) {}
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
final TcpOutboundGateway gateway = new TcpOutboundGateway();
gateway.setConnectionFactory(ccf);
gateway.setRequestTimeout(Integer.MAX_VALUE);
QueueChannel replyChannel = new QueueChannel();
gateway.setRequiresReply(true);
gateway.setOutputChannel(replyChannel);
gateway.setRemoteTimeoutExpression(new SpelExpressionParser().parseExpression("5000"));
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.afterPropertiesSet();
gateway.start();
try {
gateway.handleMessage(MessageBuilder.withPayload("Test").build());
fail("expected failure");
}
catch (Exception e) {
assertThat(e.getCause(), instanceOf(SocketTimeoutException.class));
}
assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size());
Message<?> reply = replyChannel.receive(0);
assertNull(reply);
done.set(true);
ccf.getConnection();
gateway.stop();
ccf.stop();
}
}