From 0301f262e0fbef9775c1631ea16971029241a199 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 18 Aug 2013 16:59:52 -0400 Subject: [PATCH] INT-3103 TCP Propagate Exceptions to GW Thread Previously, when a calling thread was waiting for a reply, and an exception occurred on the socket, the exception was not propagated to the thread and it would eventually get a timeout, but with no indication of the problem. Propagate the exception to the calling thread by invoking onMessage() with an ErrorMessage. Ignore the ErrorMessage in other TcpListeners (inbound adapter, gateway). Ensure NIO closes are not missed by sending an ErrorMessage when the selector thread detects a closed channel. Add tests for Net, NIO, cached and failover connection factories. Polishing Remove left-over debug logs; add comment to setReply(). Add comments for ErrorMessages in onMessage(). --- .../integration/ip/tcp/TcpInboundGateway.java | 10 +- .../ip/tcp/TcpOutboundGateway.java | 51 +++++++- .../ip/tcp/TcpReceivingChannelAdapter.java | 8 ++ .../connection/AbstractConnectionFactory.java | 4 + .../TcpConnectionInterceptorSupport.java | 8 +- .../tcp/connection/TcpConnectionSupport.java | 15 +++ .../ip/tcp/connection/TcpNetConnection.java | 1 + .../ip/tcp/connection/TcpNioConnection.java | 19 ++- .../ip/tcp/TcpOutboundGatewayTests.java | 118 ++++++++++++++++++ .../tcp/connection/TcpNetConnectionTests.java | 6 +- 10 files changed, 226 insertions(+), 14 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java index 5e54612689..1309caed33 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -32,6 +32,7 @@ import org.springframework.integration.ip.tcp.connection.ClientModeConnectionMan import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -53,7 +54,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements private volatile AbstractClientConnectionFactory clientConnectionFactory; - private Map connections = new ConcurrentHashMap(); + private final Map connections = new ConcurrentHashMap(); private volatile boolean isClientMode; @@ -76,6 +77,13 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements } } else { + if (message instanceof ErrorMessage) { + /* + * Socket errors are sent here so they can be conveyed to any waiting thread. + * There's not one here; simply ignore. + */ + return false; + } this.activeCount.incrementAndGet(); try { return doOnMessage(message); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index cdf408938e..1f0d005e98 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -33,7 +33,9 @@ import org.springframework.integration.ip.tcp.connection.AbstractClientConnectio import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -163,8 +165,17 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp } AsyncReply reply = pendingReplies.get(connectionId); if (reply == null) { - logger.error("Cannot correlate response - no pending reply"); - return false; + if (message instanceof ErrorMessage) { + /* + * Socket errors are sent here so they can be conveyed to any waiting thread. + * If there's not one, simply ignore. + */ + return false; + } + else { + logger.error("Cannot correlate response - no pending reply"); + return false; + } } reply.setReply(message); return false; @@ -248,10 +259,13 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp private final CountDownLatch latch; + private final CountDownLatch secondChanceLatch; + private volatile Message reply; public AsyncReply() { this.latch = new CountDownLatch(1); + this.secondChanceLatch = new CountDownLatch(1); } /** @@ -268,12 +282,41 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp catch (InterruptedException e) { Thread.currentThread().interrupt(); } + boolean waitForMessageAfterError = connectionFactory instanceof TcpNioClientConnectionFactory; + while (reply instanceof ErrorMessage) { + if (waitForMessageAfterError) { + /* + * Possible race condition with NIO; we might have received the close + * before the reply, on a different thread. + */ + logger.debug("second chance"); + this.secondChanceLatch.await(2, TimeUnit.SECONDS); + waitForMessageAfterError = false; + } + else if (reply.getPayload() instanceof MessagingException) { + throw (MessagingException) reply.getPayload(); + } + else { + throw new MessagingException("Exception while awaiting reply", (Throwable) reply.getPayload()); + } + } return this.reply; } + /** + * We have a race condition when a socket is closed right after the reply is received. The close "error" + * might arrive before the actual reply. Overwrite an error with a good reply, but not vice-versa. + * @param reply + */ public void setReply(Message reply) { - this.reply = reply; - this.latch.countDown(); + if (this.reply == null) { + this.reply = reply; + this.latch.countDown(); + } + else if (this.reply instanceof ErrorMessage) { + this.reply = reply; + this.secondChanceLatch.countDown(); + } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index 6b0506c229..61e607dcbb 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -28,6 +28,7 @@ import org.springframework.integration.ip.tcp.connection.ClientModeCapable; import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager; import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -68,6 +69,13 @@ public class TcpReceivingChannelAdapter } } else { + if (message instanceof ErrorMessage) { + /* + * Socket errors are sent here so they can be conveyed to any waiting thread. + * There's not one here; simply ignore. + */ + return false; + } this.activeCount.incrementAndGet(); try { sendMessage(message); 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 25b11d562c..251d952c41 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 @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.EOFException; import java.io.IOException; import java.net.Socket; import java.net.SocketException; @@ -597,6 +598,9 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport key.interestOps(SelectionKey.OP_READ); selector.wakeup(); } + else { + connection.sendExceptionToListener(new EOFException("Connection is closed")); + } }}); } else if (key.isAcceptable()) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java index 3a988fa4cc..e46ef42ec5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java @@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; +import org.springframework.integration.message.ErrorMessage; /** * Base class for TcpConnectionIntercepters; passes all method calls through @@ -131,7 +132,12 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo public boolean onMessage(Message message) { if (this.tcpListener == null) { - throw new NoListenerException("No listener registered for message reception"); + if (message instanceof ErrorMessage) { + return false; + } + else { + throw new NoListenerException("No listener registered for message reception"); + } } return this.tcpListener.onMessage(message); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index f2c08d1b75..2b702091b6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -19,6 +19,8 @@ package org.springframework.integration.ip.tcp.connection; import java.net.InetAddress; import java.net.Socket; import java.net.SocketException; +import java.util.Collections; +import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -30,7 +32,9 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; +import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -80,6 +84,8 @@ public abstract class TcpConnectionSupport implements TcpConnection { private final AtomicBoolean closePublished = new AtomicBoolean(); + private final AtomicBoolean exceptionSent = new AtomicBoolean(); + public TcpConnectionSupport() { this.server = false; this.applicationEventPublisher = null; @@ -302,6 +308,15 @@ public abstract class TcpConnectionSupport implements TcpConnection { return this.connectionId; } + protected final void sendExceptionToListener(Exception e) { + if (!this.exceptionSent.getAndSet(true) && this.getListener() != null) { + Map headers = Collections.singletonMap(IpHeaders.CONNECTION_ID, + (Object) this.getConnectionId()); + ErrorMessage errorMessage = new ErrorMessage(e, headers); + this.getListener().onMessage(errorMessage); + } + } + protected void publishConnectionOpenEvent() { TcpConnectionEvent event = new TcpConnectionOpenEvent(this, this.connectionFactoryName); 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 2472f9a747..fc5821d353 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 @@ -248,6 +248,7 @@ public class TcpNetConnection extends TcpConnectionSupport { e.getClass().getSimpleName() + ":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage()); } + this.sendExceptionToListener(e); } } } 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 a4bf9a7f5f..75f2d3b455 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 @@ -201,23 +201,28 @@ public class TcpNioConnection extends TcpConnectionSupport { if (message != null) { sendToChannel(message); } - } else { + } + else { this.executionControl.decrementAndGet(); } - } catch (Exception e) { + } + catch (Exception e) { if (logger.isTraceEnabled()) { logger.error("Read exception " + this.getConnectionId(), e); - } else { + } + else { logger.error("Read exception " + this.getConnectionId() + " " + e.getClass().getSimpleName() + ":" + e.getCause() + ":" + e.getMessage()); } this.closeConnection(); + this.sendExceptionToListener(e); return; } - } finally { + } + finally { if (logger.isTraceEnabled()) { logger.trace(this.getConnectionId() + " Nio message assembler exiting..."); } @@ -228,7 +233,8 @@ public class TcpNioConnection extends TcpConnectionSupport { if (this.isOpen() && dataAvailable()) { checkForAssembler(); } - } catch (IOException e) { + } + catch (IOException e) { logger.error("Exception when checking for assembler", e); } } @@ -293,7 +299,7 @@ public class TcpNioConnection extends TcpConnectionSupport { */ if (this.isSingleUse() && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) { if (logger.isDebugEnabled()) { - logger.debug("Closing single use cbannel after inbound message " + this.getConnectionId()); + logger.debug("Closing single use channel after inbound message " + this.getConnectionId()); } this.closeConnection(); } @@ -377,6 +383,7 @@ public class TcpNioConnection extends TcpConnectionSupport { // only execute run() if we don't already have one running this.executionControl.set(1); this.taskExecutor.execute(this); + logger.debug("Running an assembler"); } else { this.executionControl.decrementAndGet(); } 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 b386809392..518d611e74 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 @@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.doThrow; @@ -25,12 +26,14 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.EOFException; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.net.ServerSocket; import java.net.Socket; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -63,6 +66,7 @@ import org.springframework.integration.ip.tcp.connection.CachingClientConnection import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.SocketUtils; @@ -87,6 +91,7 @@ public class TcpOutboundGatewayTests { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100); latch.countDown(); + List sockets = new ArrayList(); int i = 0; while (true) { Socket socket = server.accept(); @@ -94,6 +99,7 @@ public class TcpOutboundGatewayTests { ois.readObject(); ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); oos.writeObject("Reply" + (i++)); + sockets.add(socket); } } catch (Exception e) { if (!done.get()) { @@ -568,4 +574,116 @@ public class TcpOutboundGatewayTests { return connection; } + @Test + public void testNetGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + ccf.start(); + testGWPropagatesSocketCloseGuts(port, ccf); + } + + @Test + public void testNioGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + ccf.start(); + testGWPropagatesSocketCloseGuts(port, ccf); + } + + @Test + public void testCachedGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(ccf, 1); + cccf.start(); + testGWPropagatesSocketCloseGuts(port, cccf); + } + + @Test + public void testFailoverGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + FailoverClientConnectionFactory focf = new FailoverClientConnectionFactory( + Collections.singletonList(ccf)); + focf.start(); + testGWPropagatesSocketCloseGuts(port, focf); + } + + private void testGWPropagatesSocketCloseGuts(final int port, AbstractClientConnectionFactory ccf) throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + final AtomicReference lastReceived = new AtomicReference(); + final CountDownLatch serverLatch = new CountDownLatch(1); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + while (!done.get()) { + Socket socket = server.accept(); + i++; + while (!socket.isClosed()) { + try { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + String request = (String) ois.readObject(); + logger.debug("Read " + request + " closing socket"); + socket.close(); + lastReceived.set(request); + serverLatch.countDown(); + } + catch (IOException e) { + socket.close(); + } + } + } + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + 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) { + assertTrue(e.getCause() instanceof EOFException); + } + assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size()); + Message reply = replyChannel.receive(0); + assertNull(reply); + done.set(true); + ccf.getConnection(); + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java index d786b7a0a5..c0269c24e1 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java @@ -30,7 +30,6 @@ import java.nio.channels.SocketChannel; import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; - import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; @@ -43,6 +42,7 @@ import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream; import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer; +import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.converter.MapMessageConverter; import org.springframework.integration.test.util.TestUtils; @@ -133,7 +133,9 @@ public class TcpNetConnectionTests { TcpListener listener = new TcpListener() { public boolean onMessage(Message message) { - inboundMessage.set(message); + if (!(message instanceof ErrorMessage)) { + inboundMessage.set(message); + } return false; } };