From d395e38890e5c05edf54d5215fcc08dafc6988a3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 28 Sep 2015 10:11:23 -0400 Subject: [PATCH] INT-3837: TCP GW - Propagate Socket Timeout 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. --- .../connection/AbstractConnectionFactory.java | 4 +- .../ip/tcp/connection/TcpNetConnection.java | 2 +- .../ip/tcp/TcpOutboundGatewayTests.java | 119 +++++++++++++++++- 3 files changed, 122 insertions(+), 3 deletions(-) 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 2f3634a453..b508cd3d03 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 @@ -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); } } } 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 f30d66a882..d4cccfd847 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 @@ -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; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index d25585c6a3..215e23987a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -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 sockets = new ArrayList(); 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 sockets = new ArrayList(); + 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(); } }