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).

Conflicts:
	spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java

Resolved.

Remove beanfactory from backported test.
This commit is contained in:
Gary Russell
2015-09-28 10:11:23 -04:00
parent 14ccd6ef5b
commit d395e38890
3 changed files with 122 additions and 3 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

@@ -582,8 +582,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

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

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

@@ -16,9 +16,11 @@
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;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doThrow;
@@ -32,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;
@@ -648,11 +651,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());
@@ -673,6 +678,12 @@ public class TcpOutboundGatewayTests {
e.printStackTrace();
}
}
for (Socket socket : sockets) {
try {
socket.close();
}
catch (IOException e) {}
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
@@ -690,12 +701,118 @@ public class TcpOutboundGatewayTests {
fail("expected failure");
}
catch (Exception e) {
assertTrue(e.getCause() instanceof EOFException);
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.setRemoteTimeout(5000);
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();
}
}