INT-3745: More TCP Events (Failed Correlation)

JIRA: https://jira.spring.io/browse/INT-3745

Emit events whenever a message is received that can't be correlated to
a connection (or request in the case of an outbound gateway).

Also, when sending messages, wrap the exception in a `MessagingException` so that
the `TcpConnectionExceptionEvent` provides access to the failed message.

INT-3745: Polishing - PR Comments

Fix JavaDocs errors

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java

@@ -20,8 +20,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.springframework.messaging.Message;
-import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.integration.context.OrderlyShutdownCapable;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.integration.ip.IpHeaders;
@@ -31,8 +30,12 @@ import org.springframework.integration.ip.tcp.connection.AbstractServerConnectio
 import org.springframework.integration.ip.tcp.connection.ClientModeCapable;
 import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager;
 import org.springframework.integration.ip.tcp.connection.TcpConnection;
+import org.springframework.integration.ip.tcp.connection.TcpConnectionFailedCorrelationEvent;
 import org.springframework.integration.ip.tcp.connection.TcpListener;
 import org.springframework.integration.ip.tcp.connection.TcpSender;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
 
 /**
@@ -70,6 +73,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 
 	private final AtomicInteger activeCount = new AtomicInteger();
 
+	@Override
 	public boolean onMessage(Message<?> message) {
 		if (this.shuttingDown) {
 			if (logger.isInfoEnabled()) {
@@ -109,17 +113,29 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 			connection = connections.get(connectionId);
 		}
 		if (connection == null) {
+			publishNoConnectionEvent(message, connectionId);
 			logger.error("Connection not found when processing reply " + reply + " for " + message);
 			return false;
 		}
 		try {
 			connection.send(reply);
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			logger.error("Failed to send reply", e);
 		}
 		return false;
 	}
 
+	private void publishNoConnectionEvent(Message<?> message, String connectionId) {
+		AbstractConnectionFactory cf = this.serverConnectionFactory != null ? this.serverConnectionFactory
+				: this.clientConnectionFactory;
+		ApplicationEventPublisher applicationEventPublisher = cf.getApplicationEventPublisher();
+		if (applicationEventPublisher != null) {
+			applicationEventPublisher.publishEvent(
+				new TcpConnectionFailedCorrelationEvent(this, connectionId, new MessagingException(message)));
+		}
+	}
+
 	/**
 	 * @return true if the associated connection factory is listening.
 	 */
@@ -147,10 +163,12 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 		connectionFactory.registerSender(this);
 	}
 
+	@Override
 	public void addNewConnection(TcpConnection connection) {
 		connections.put(connection.getConnectionId(), connection);
 	}
 
+	@Override
 	public void removeDeadConnection(TcpConnection connection) {
 		connections.remove(connection.getConnectionId());
 	}
@@ -213,6 +231,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 	/**
 	 * @return the isClientMode
 	 */
+	@Override
 	public boolean isClientMode() {
 		return isClientMode;
 	}
@@ -240,6 +259,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 		this.retryInterval = retryInterval;
 	}
 
+	@Override
 	public boolean isClientModeConnected() {
 		if (this.isClientMode && this.clientModeConnectionManager != null) {
 			return this.clientModeConnectionManager.isConnected();
@@ -248,17 +268,20 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
 		}
 	}
 
+	@Override
 	public void retryConnection() {
 		if (this.active && this.isClientMode && this.clientModeConnectionManager != null) {
 			this.clientModeConnectionManager.run();
 		}
 	}
 
+	@Override
 	public int beforeShutdown() {
 		this.shuttingDown = true;
 		return this.activeCount.get();
 	}
 
+	@Override
 	public int afterShutdown() {
 		this.stop();
 		return this.activeCount.get();

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java

@@ -22,6 +22,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.Lifecycle;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.expression.EvaluationContext;
@@ -36,6 +37,7 @@ import org.springframework.integration.ip.tcp.connection.AbstractClientConnectio
 import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.CloseDeferrable;
 import org.springframework.integration.ip.tcp.connection.TcpConnection;
+import org.springframework.integration.ip.tcp.connection.TcpConnectionFailedCorrelationEvent;
 import org.springframework.integration.ip.tcp.connection.TcpListener;
 import org.springframework.integration.ip.tcp.connection.TcpSender;
 import org.springframework.messaging.Message;
@@ -167,6 +169,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 		String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID);
 		if (connectionId == null) {
 			logger.error("Cannot correlate response - no connection id");
+			publishNoConnectionEvent(message, null, "Cannot correlate response - no connection id");
 			return false;
 		}
 		if (logger.isTraceEnabled()) {
@@ -182,7 +185,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 				return false;
 			}
 			else {
-				logger.error("Cannot correlate response - no pending reply");
+				String errorMessage = "Cannot correlate response - no pending reply for " + connectionId;
+				logger.error(errorMessage);
+				publishNoConnectionEvent(message, connectionId, errorMessage);
 				return false;
 			}
 		}
@@ -190,6 +195,14 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
 		return false;
 	}
 
+	private void publishNoConnectionEvent(Message<?> message, String connectionId, String errorMessage) {
+		ApplicationEventPublisher applicationEventPublisher = this.connectionFactory.getApplicationEventPublisher();
+		if (applicationEventPublisher != null) {
+			applicationEventPublisher.publishEvent(new TcpConnectionFailedCorrelationEvent(this, connectionId,
+					new MessagingException(message, errorMessage)));
+		}
+	}
+
 	public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
 		// TODO: In 3.0 Change parameter type to AbstractClientConnectionFactory
 		Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory,

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java

@@ -20,8 +20,8 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledFuture;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.Lifecycle;
-import org.springframework.messaging.MessageHandlingException;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -30,8 +30,10 @@ 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.TcpConnection;
+import org.springframework.integration.ip.tcp.connection.TcpConnectionFailedCorrelationEvent;
 import org.springframework.integration.ip.tcp.connection.TcpSender;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 
@@ -110,7 +112,10 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 			}
 			else {
 				logger.error("Unable to find outbound socket for " + message);
-				throw new MessageHandlingException(message, "Unable to find outbound socket");
+				MessageHandlingException messageHandlingException = new MessageHandlingException(message,
+						"Unable to find outbound socket");
+				publishNoConnectionEvent(messageHandlingException, (String) connectionId);
+				throw messageHandlingException;
 			}
 			return;
 		}
@@ -159,6 +164,16 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
+	private void publishNoConnectionEvent(MessageHandlingException messageHandlingException, String connectionId) {
+		AbstractConnectionFactory cf = this.serverConnectionFactory != null ? this.serverConnectionFactory
+				: this.clientConnectionFactory;
+		ApplicationEventPublisher applicationEventPublisher = cf.getApplicationEventPublisher();
+		if (applicationEventPublisher != null) {
+			applicationEventPublisher.publishEvent(
+				new TcpConnectionFailedCorrelationEvent(this, connectionId, messageHandlingException));
+		}
+	}
+
 	/**
 	 * Sets the client or server connection factory; for this (an outbound adapter), if
 	 * the factory is a server connection factory, the sockets are owned by a receiving
@@ -175,10 +190,12 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
+	@Override
 	public void addNewConnection(TcpConnection connection) {
 		connections.put(connection.getConnectionId(), connection);
 	}
 
+	@Override
 	public void removeDeadConnection(TcpConnection connection) {
 		connections.remove(connection.getConnectionId());
 	}
@@ -199,6 +216,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
+	@Override
 	public void start() {
 		synchronized (this.lifecycleMonitor) {
 			if (!this.active) {
@@ -220,6 +238,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
+	@Override
 	public void stop() {
 		synchronized (this.lifecycleMonitor) {
 			if (this.active) {
@@ -237,6 +256,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
+	@Override
 	public boolean isRunning() {
 		return this.active;
 	}
@@ -283,6 +303,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 	/**
 	 * @return the isClientMode
 	 */
+	@Override
 	public boolean isClientMode() {
 		return this.isClientMode;
 	}
@@ -315,6 +336,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		this.retryInterval = retryInterval;
 	}
 
+	@Override
 	public boolean isClientModeConnected() {
 		if (this.isClientMode && this.clientModeConnectionManager != null) {
 			return this.clientModeConnectionManager.isConnected();
@@ -323,6 +345,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
 		}
 	}
 
+	@Override
 	public void retryConnection() {
 		if (this.active && this.isClientMode && this.clientModeConnectionManager != null) {
 			this.clientModeConnectionManager.run();

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

@@ -147,7 +147,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 		}
 	}
 
-	protected ApplicationEventPublisher getApplicationEventPublisher() {
+	public ApplicationEventPublisher getApplicationEventPublisher() {
 		return applicationEventPublisher;
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionFailedCorrelationEvent.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ip.tcp.connection;
+
+import org.springframework.integration.ip.event.IpIntegrationEvent;
+import org.springframework.messaging.MessagingException;
+
+
+
+/**
+ * An event emitted when an endpoint cannot correlate a connection id to a
+ * connection; the cause is a messaging exception with the failed message.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class TcpConnectionFailedCorrelationEvent extends IpIntegrationEvent {
+
+	private static final long serialVersionUID = -7460880274740273542L;
+
+	private final String connectionId;
+
+	public TcpConnectionFailedCorrelationEvent(Object source, String connectionId, MessagingException cause) {
+		super(source, cause);
+		this.connectionId = connectionId;
+	}
+
+	public String getConnectionId() {
+		return connectionId;
+	}
+
+	@Override
+	public String toString() {
+		return super.toString() +
+				", [connectionId=" + this.connectionId + "]";
+	}
+
+}

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

@@ -30,6 +30,7 @@ import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 
 /**
@@ -105,7 +106,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
 			this.socketOutputStream.flush();
 		}
 		catch (Exception e) {
-			this.publishConnectionExceptionEvent(e);
+			this.publishConnectionExceptionEvent(new MessagingException(message, e));
 			this.closeConnection(true);
 			throw e;
 		}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java

@@ -43,6 +43,7 @@ import org.springframework.core.serializer.Serializer;
 import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
 import org.springframework.integration.util.CompositeExecutor;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 
 /**
@@ -150,7 +151,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 				this.bufferedOutputStream.flush();
 			}
 			catch (Exception e) {
-				this.publishConnectionExceptionEvent(e);
+				this.publishConnectionExceptionEvent(new MessagingException(message, e));
 				this.closeConnection(true);
 				throw e;
 			}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java

@@ -20,7 +20,9 @@ import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.startsWith;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
@@ -46,6 +48,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ServerSocketFactory;
 
 import org.apache.commons.logging.Log;
+import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
@@ -55,9 +58,19 @@ import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.serializer.Serializer;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.ip.tcp.TcpInboundGateway;
+import org.springframework.integration.ip.tcp.TcpOutboundGateway;
+import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.SocketUtils;
 
 /**
@@ -103,10 +116,10 @@ public class ConnectionEventTests {
 		assertNotNull(theEvent.get(0));
 		assertTrue(theEvent.get(0) instanceof TcpConnectionExceptionEvent);
 		assertTrue(theEvent.get(0).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]"));
-		assertTrue(theEvent.get(0).toString().contains("cause=java.lang.RuntimeException: foo]"));
+		assertThat(theEvent.get(0).toString(), containsString("RuntimeException: foo]"));
 		TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(0);
 		assertNotNull(event.getCause());
-		assertSame(toBeThrown, event.getCause());
+		assertSame(toBeThrown, event.getCause().getCause());
 		assertTrue(theEvent.size() > 1);
 		assertNotNull(theEvent.get(1));
 		assertTrue(theEvent.get(1).toString()
@@ -127,6 +140,138 @@ public class ConnectionEventTests {
 		testServerExceptionGuts(port, factory);
 	}
 
+	@Test
+	public void testOutboundChannelAdapterNoConnectionEvents() {
+		TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
+		AbstractServerConnectionFactory scf = new AbstractServerConnectionFactory(0) {
+
+			@Override
+			public void run() {
+			}
+		};
+		final AtomicReference<ApplicationEvent> theEvent = new AtomicReference<ApplicationEvent>();
+		scf.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(Object event) {
+			}
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				theEvent.set(event);
+			}
+
+		});
+		handler.setConnectionFactory(scf);
+		handler.start();
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader(IpHeaders.CONNECTION_ID, "bar")
+				.build();
+		try {
+			handler.handleMessage(message);
+			fail("expected exception");
+		}
+		catch (MessageHandlingException e) {
+			assertThat(e.getMessage(), Matchers.containsString("Unable to find outbound socket"));
+		}
+		assertNotNull(theEvent.get());
+		TcpConnectionFailedCorrelationEvent event = (TcpConnectionFailedCorrelationEvent) theEvent.get();
+		assertEquals("bar", event.getConnectionId());
+		assertSame(message, ((MessagingException) event.getCause()).getFailedMessage());
+	}
+
+	@Test
+	public void testInboundGatewayNoConnectionEvents() {
+		TcpInboundGateway gw = new TcpInboundGateway();
+		AbstractServerConnectionFactory scf = new AbstractServerConnectionFactory(0) {
+
+			@Override
+			public void run() {
+			}
+		};
+		final AtomicReference<ApplicationEvent> theEvent = new AtomicReference<ApplicationEvent>();
+		scf.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(Object event) {
+			}
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				theEvent.set(event);
+			}
+
+		});
+		gw.setConnectionFactory(scf);
+		DirectChannel requestChannel = new DirectChannel();
+		requestChannel.subscribe(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				((MessageChannel) message.getHeaders().getReplyChannel()).send(message);
+			}
+		});
+		gw.setRequestChannel(requestChannel);
+		gw.start();
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader(IpHeaders.CONNECTION_ID, "bar")
+				.build();
+		gw.onMessage(message);
+		assertNotNull(theEvent.get());
+		TcpConnectionFailedCorrelationEvent event = (TcpConnectionFailedCorrelationEvent) theEvent.get();
+		assertEquals("bar", event.getConnectionId());
+		assertSame(message, ((MessagingException) event.getCause()).getFailedMessage());
+	}
+
+	@Test
+	public void testOutboundGatewayNoConnectionEvents() {
+		TcpOutboundGateway gw = new TcpOutboundGateway();
+		AbstractClientConnectionFactory ccf = new AbstractClientConnectionFactory("localhost", 0) {
+		};
+		final AtomicReference<ApplicationEvent> theEvent = new AtomicReference<ApplicationEvent>();
+		ccf.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(Object event) {
+			}
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				theEvent.set(event);
+			}
+
+		});
+		gw.setConnectionFactory(ccf);
+		DirectChannel requestChannel = new DirectChannel();
+		requestChannel.subscribe(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				((MessageChannel) message.getHeaders().getReplyChannel()).send(message);
+			}
+		});
+		gw.start();
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader(IpHeaders.CONNECTION_ID, "bar")
+				.build();
+		gw.onMessage(message);
+		assertNotNull(theEvent.get());
+		TcpConnectionFailedCorrelationEvent event = (TcpConnectionFailedCorrelationEvent) theEvent.get();
+		assertEquals("bar", event.getConnectionId());
+		MessagingException messagingException = (MessagingException) event.getCause();
+		assertSame(message, messagingException.getFailedMessage());
+		assertEquals("Cannot correlate response - no pending reply for bar", messagingException.getMessage());
+
+		message = new GenericMessage<String>("foo");
+		gw.onMessage(message);
+		assertNotNull(theEvent.get());
+		event = (TcpConnectionFailedCorrelationEvent) theEvent.get();
+		assertNull(event.getConnectionId());
+		messagingException = (MessagingException) event.getCause();
+		assertSame(message, messagingException.getFailedMessage());
+		assertEquals("Cannot correlate response - no connection id", messagingException.getMessage());
+	}
+
 	private void testServerExceptionGuts(int port, AbstractServerConnectionFactory factory) throws Exception {
 		ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(port);
 		final AtomicReference<TcpConnectionServerExceptionEvent> theEvent =