From 9e51ba4ea4688f6e22585b7770df30015f87eead 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`. --- .../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 980121a1b6..f61b4cec5a 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 9ea8857192..12af25af0e 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);