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

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<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
+	private final Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
 
 	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);

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

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

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()) {

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

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<String, Object> 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);

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

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

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<Socket> sockets = new ArrayList<Socket>();
 					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<String> lastReceived = new AtomicReference<String>();
+		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();
+	}
 }

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;
 			}
 		};