Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4024: TcpNetConnection Do Not Cache Listener

Browse Source 
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)
This commit is contained in:
Gary Russell
2016-05-10 15:58:36 -04:00
committed by Artem Bilan
parent 366d7ab2a1
commit 2fa55e7acb
2 changed files with 57 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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");
}

56
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java
View File

@@ -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<AbstractClientConnectionFactory> clientFactories = new ArrayList<AbstractClientConnectionFactory>();
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<String>("foo"));
Message<byte[]> result = (Message<byte[]>) replyChannel.receive(10000);
assertNotNull(result);
assertEquals("foo", new String(result.getPayload()));
// INT-4024 - second reply had bad connection id
outbound.handleMessage(new GenericMessage<String>("foo"));
result = (Message<byte[]>) 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);