From f4104c199a2aefd338762598f7bbda1cae17fbb9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 17 Jun 2015 13:38:11 -0400 Subject: [PATCH] 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 --- .../AbstractRemoteFileOutboundGateway.java | 2 +- .../integration/ip/tcp/TcpInboundGateway.java | 29 +++- .../ip/tcp/TcpOutboundGateway.java | 15 +- .../ip/tcp/TcpSendingMessageHandler.java | 27 +++- .../connection/AbstractConnectionFactory.java | 2 +- .../TcpConnectionFailedCorrelationEvent.java | 52 ++++++ .../ip/tcp/connection/TcpNetConnection.java | 3 +- .../ip/tcp/connection/TcpNioConnection.java | 3 +- .../tcp/connection/ConnectionEventTests.java | 149 +++++++++++++++++- src/reference/asciidoc/ip.adoc | 8 +- src/reference/asciidoc/whats-new.adoc | 22 +-- 11 files changed, 289 insertions(+), 23 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionFailedCorrelationEvent.java diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index 1549aea010..6133bd24b4 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -175,7 +175,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply RECURSIVE("-R"), /** - * Streaming 'get' (returns InputStream); user must call {@link RemoteFileUtils#closeSession(Session)}. + * Streaming 'get' (returns InputStream); user must call {@link Session#close()}. */ STREAM("-stream"); 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 09a7454c9d..23bb903b55 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 @@ -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(); 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 bbe785b57b..a360a121e1 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 @@ -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, diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index e2647529ad..4f31826c58 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/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(); 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 55bf5bc176..fe89eaf3c8 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 @@ -147,7 +147,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport } } - protected ApplicationEventPublisher getApplicationEventPublisher() { + public ApplicationEventPublisher getApplicationEventPublisher() { return applicationEventPublisher; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionFailedCorrelationEvent.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionFailedCorrelationEvent.java new file mode 100644 index 0000000000..6776cdcdfb --- /dev/null +++ b/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 + "]"; + } + +} 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 d6f5708235..9bcc4eaa0b 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 @@ -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; } 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 40200c7974..ae01358b12 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 @@ -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; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java index 7858a63255..5aa07e89f1 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java +++ b/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 theEvent = new AtomicReference(); + 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 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 theEvent = new AtomicReference(); + 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 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 theEvent = new AtomicReference(); + 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 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("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 theEvent = diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 12c64de3d9..6246962a81 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -373,9 +373,15 @@ In addition, since _version 4.0_ the standard deserializers discussed in <>, to capture these events, allowing analysis of the problem. -Starting with _versions 4.0.7, 4.1.3_`TcpConnectionServerExceptionEvent` s are published whenever an unexpected exception occurs on a server socket (such as a `BindException` when the server socket is in use). +Starting with _versions 4.0.7, 4.1.3_, `TcpConnectionServerExceptionEvent` s are published whenever an unexpected exception occurs on a server socket (such as a `BindException` when the server socket is in use). These events have a reference to the connection factory and the cause. +Starting with _version 4.2_, `TcpConnectionFailedCorrelationEvent` s are published whenever an endpoint (inbound gateway or +collaborating outbound channel adapter) receives a message that cannot be routed to a connection because the +`ip_connectionId` header is invalid. +Outbound gateways also publish this event when a late reply is received (the sender thread has timed out). +The event contains the connection id as well as an exception in the `cause` property that contains the failed message. + [[tcp-adapters]] === TCP Adapters diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 6c383f7cab..41507fa99d 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -63,20 +63,22 @@ See <> for more information. The `ScatterGatherHandler` class has been moved from the `org.springframework.integration.handler` to the `org.springframework.integration.scattergather`. +==== TCP Changes + [[x4.2-tcp-serializers]] -==== TCP Serializers +===== TCP Serializers The TCP `Serializers` no longer `flush()` the `OutputStream`; this is now done by the `TcpNxxConnection` classes. If you are using the serializers directly within user code, you may have to `flush()` the `OutputStream`. [[x4.2-tcp-server-exceptions]] -==== Server Socket Exceptions +===== Server Socket Exceptions `TcpConnectionServerExceptionEvent` s are now published whenever an unexpected exception occurs on a TCP server socket (also added to 4.1.3, 4.0.7). See <> for more information. [[x4.2-tcp-gw-rto]] -==== TCP Gateway Remote Timeout +===== TCP Gateway Remote Timeout The `TcpOutboundGateway` now supports `remote-timeout-expression` as an alternative to the existing `remote-timeout` attribute. This allows setting the timeout based on each message. @@ -86,7 +88,7 @@ Also, the `remote-timeout` no longer defaults to the same value as `reply-timeou See <> for more information. [[x4.2-tcp-ssl]] -==== TCP SSLSession Available for Header Mapping +===== TCP SSLSession Available for Header Mapping `TcpConnection` s now support `getSslSession()` to enable users to extract information from the session to add to message headers. @@ -94,6 +96,12 @@ message headers. See <> for more information. +[[x4.2-tcp-events]] +===== TCP Events + +New events are now published whenever a correlation exception occurs - for example sending a message to a +non-existent socket. See <> for more information. + [[x4.2-inbound-channel-adapter-annotation]] ==== @InboundChannelAdapter @@ -160,9 +168,3 @@ metadata store if it implements `Flushable` (e.g. the `PropertiesPersistenMetada When using Java 8, gateway methods can now return `CompletableFuture`. See <> for more information. - -[[x4.2-ftp-changes]] -==== (S)FTP Changes - -The SFTP/FTP outbound gateway `get` operations now support the `-stream` option, allowing files to be retrieved as -a stream. See <> and <> for more information.