From 2fa55e7acbd28dbc136c1b9b4d244ed779799553 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 10 May 2016 15:58:36 -0400 Subject: [PATCH] INT-4024: TcpNetConnection Do Not Cache Listener JIRA; https://jira.spring.io/browse/INT-4024 When the `FailOverClientConnectionFactory` is used on top of a `CachingClientConnectionFactory`, the underlying connection's listener is changed to the new `CachedConnection` which in turn has its listener set to the new `FailoverTcpConnection`. Since the caching connection factory implies single-use (so the connection is "closed" - returned to the cache), the `FOCCF` is also forced to have single-use = true - hence the new ultimate listener. The `TcpNetConnection` obtained its listener (`getListner()`) in the outer block of its read loop and hence did not detect the listener change. This caused the old listener (cached connection) to be invoked and when the result was ultimately passed back to the `TcpOutboundGateway`, the connection id did not match and the reply eventually timed out. Change the `TcpNetConnection` to call `getListener()` for every message. This was already the case in `TcpNioConnection`. (cherry picked from commit 9e51ba4) --- .../ip/tcp/connection/TcpNetConnection.java | 2 +- .../FailoverClientConnectionFactoryTests.java | 56 +++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) 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 db73960502..9d65239edd 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 @@ -154,7 +154,6 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling */ @Override public void run() { - TcpListener listener = getListener(); boolean okToRun = true; if (logger.isDebugEnabled()) { logger.debug(this.getConnectionId() + " Reading..."); @@ -176,6 +175,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling logger.debug("Message received " + message); } try { + TcpListener listener = getListener(); if (listener == null) { throw new NoListenerException("No listener"); } 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 682f0da848..f1fc95128f 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 @@ -53,6 +53,7 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.TcpInboundGateway; import org.springframework.integration.ip.tcp.TcpOutboundGateway; @@ -405,6 +406,61 @@ public class FailoverClientConnectionFactoryTests { server2.stop(); } + @SuppressWarnings("unchecked") + @Test + public void testFailoverCachedWithGateway() throws Exception { + final TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0); + server.setBeanName("server"); + server.afterPropertiesSet(); + DirectChannel inChannel = new DirectChannel(); + inChannel.setBeanName("inChannel"); + TcpInboundGateway inbound = new TcpInboundGateway(); + inbound.setConnectionFactory(server); + inbound.setRequestChannel(inChannel); + inbound.afterPropertiesSet(); + inChannel.subscribe(new BridgeHandler()); + inbound.start(); + TestingUtilities.waitListening(server, 10000L); + int port = server.getPort(); + AbstractClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port); + client.setBeanName("client"); + + // Cache + CachingClientConnectionFactory cachingClient = new CachingClientConnectionFactory(client, 2); + cachingClient.setBeanName("cache"); + cachingClient.afterPropertiesSet(); + + // Failover + List clientFactories = new ArrayList(); + clientFactories.add(cachingClient); + FailoverClientConnectionFactory failoverClient = new FailoverClientConnectionFactory(clientFactories); + failoverClient.setSingleUse(true); + failoverClient.afterPropertiesSet(); + + TcpOutboundGateway outbound = new TcpOutboundGateway(); + outbound.setConnectionFactory(failoverClient); + QueueChannel replyChannel = new QueueChannel(); + replyChannel.setBeanName("replyChannel"); + outbound.setReplyChannel(replyChannel); + outbound.setBeanFactory(mock(BeanFactory.class)); + outbound.afterPropertiesSet(); + outbound.start(); + + outbound.handleMessage(new GenericMessage("foo")); + Message result = (Message) replyChannel.receive(10000); + assertNotNull(result); + assertEquals("foo", new String(result.getPayload())); + + // INT-4024 - second reply had bad connection id + outbound.handleMessage(new GenericMessage("foo")); + result = (Message) replyChannel.receive(10000); + assertNotNull(result); + assertEquals("foo", new String(result.getPayload())); + + inbound.stop(); + outbound.stop(); + } + @Test public void testFailoverCachedRealBadHost() throws Exception { TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(0);