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 1f0d005e98..be81330c89 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,6 @@ 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; @@ -138,10 +137,10 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp return replyMessage; } catch (Exception e) { + logger.error("Tcp Gateway exception", e); if (e instanceof MessagingException) { throw (MessagingException) e; } - logger.error("Tcp Gateway exception", e); throw new MessagingException("Failed to send or receive", e); } finally { @@ -163,6 +162,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp logger.error("Cannot correlate response - no connection id"); return false; } + if (logger.isTraceEnabled()) { + logger.trace("onMessage: " + connectionId + "(" + message + ")"); + } AsyncReply reply = pendingReplies.get(connectionId); if (reply == null) { if (message instanceof ErrorMessage) { @@ -282,7 +284,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp catch (InterruptedException e) { Thread.currentThread().interrupt(); } - boolean waitForMessageAfterError = connectionFactory instanceof TcpNioClientConnectionFactory; + boolean waitForMessageAfterError = true; while (reply instanceof ErrorMessage) { if (waitForMessageAfterError) { /* diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index 5889e7d34d..c552b0fe90 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -22,6 +22,7 @@ import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.util.SimplePool; @@ -317,7 +318,10 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact super.registerListener(listener); targetConnectionFactory.registerListener(new TcpListener() { public boolean onMessage(Message message) { - throw new UnsupportedOperationException("This should never be called"); + if (!(message instanceof ErrorMessage)) { + throw new UnsupportedOperationException("This should never be called"); + } + return false; } }); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java index b484b54ac9..2b75454027 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java @@ -19,11 +19,13 @@ import java.io.IOException; import java.util.Iterator; import java.util.List; import java.util.UUID; +import java.util.concurrent.atomic.AtomicLong; 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.message.ErrorMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; @@ -81,7 +83,10 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac for (AbstractClientConnectionFactory factory : this.factories) { factory.registerListener(new TcpListener() { public boolean onMessage(Message message) { - throw new UnsupportedOperationException("This should never be called"); + if (!(message instanceof ErrorMessage)) { + throw new UnsupportedOperationException("This should never be called"); + } + return false; } }); } @@ -98,10 +103,12 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac protected TcpConnectionSupport obtainConnection() throws Exception { TcpConnectionSupport connection = this.getTheConnection(); if (connection != null && connection.isOpen()) { + ((FailoverTcpConnection) connection).incrementEpoch(); return connection; } FailoverTcpConnection failoverTcpConnection = new FailoverTcpConnection(this.factories); failoverTcpConnection.registerListener(this.getListener()); + failoverTcpConnection.incrementEpoch(); return failoverTcpConnection; } @@ -162,6 +169,8 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac private volatile boolean open = true; + private final AtomicLong epoch = new AtomicLong(); + public FailoverTcpConnection(List factories) throws Exception { this.factories = factories; this.factoryIterator = factories.iterator(); @@ -169,6 +178,10 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac this.connectionId = UUID.randomUUID().toString(); } + void incrementEpoch() { + this.epoch.incrementAndGet(); + } + /** * Finds a connection from the underlying list of factories. If necessary, * each factory is tried; including the current one if we wrap around. @@ -288,7 +301,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac @Override public String getConnectionId() { - return this.connectionId; + return this.connectionId + ":" + epoch; } @Override @@ -331,11 +344,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac this.delegate.setSerializer(serializer); } - @Override - public long incrementAndGetConnectionSequence() { - return this.delegate.incrementAndGetConnectionSequence(); - } - /** * We have to intercept the message to replace the connectionId header with * ours so the listener can correlate a response with a request. We supply @@ -343,13 +351,21 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac * purposes. */ public boolean onMessage(Message message) { - MessageBuilder messageBuilder = MessageBuilder.fromMessage(message) - .setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId()); - if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) { - messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID, - message.getHeaders().get(IpHeaders.CONNECTION_ID)); + if (this.delegate.getConnectionId().equals(message.getHeaders().get(IpHeaders.CONNECTION_ID))) { + MessageBuilder messageBuilder = MessageBuilder.fromMessage(message) + .setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId()); + if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) { + messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID, + message.getHeaders().get(IpHeaders.CONNECTION_ID)); + } + return this.getListener().onMessage(messageBuilder.build()); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("Message from defunct connection ignored " + message); + } + return false; } - return this.getListener().onMessage(messageBuilder.build()); } } 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 75f2d3b455..55abdbcfa4 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 @@ -382,8 +382,10 @@ public class TcpNioConnection extends TcpConnectionSupport { if (this.executionControl.incrementAndGet() <= 1) { // only execute run() if we don't already have one running this.executionControl.set(1); + if (logger.isDebugEnabled()) { + logger.debug(this.getConnectionId() + " Running an assembler"); + } 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/connection/FailoverClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java index 9f7721b7cf..4912dc53fb 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java @@ -39,6 +39,9 @@ import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; @@ -296,6 +299,16 @@ public class FailoverClientConnectionFactoryTests { client2.setTaskExecutor(exec); server1.setTaskExecutor(exec); server2.setTaskExecutor(exec); + ApplicationEventPublisher pub = new ApplicationEventPublisher() { + + @Override + public void publishEvent(ApplicationEvent event) { + } + }; + client1.setApplicationEventPublisher(pub); + client2.setApplicationEventPublisher(pub); + server1.setApplicationEventPublisher(pub); + server2.setApplicationEventPublisher(pub); TcpInboundGateway gateway1 = new TcpInboundGateway(); gateway1.setConnectionFactory(server1); SubscribableChannel channel = new DirectChannel();