From 351af6b16b190e71092102991e6dcfa60629f179 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 18 Aug 2013 13:16:43 -0400 Subject: [PATCH] INT-3096 TCP Allow Stacking Cache and Failover Support nesting the fail over and caching client connection factories. The TcpListener chain was not set up properly; this prevented caching Failover connections, or Failover cached connections. Each nested interceptor needs the next outer interceptor set up as its listener, so the underlying connection calls onMessage() all the way up to the ultimate listener (adapter or gateway). Fix the Listener hierarchy; add tests for caching failover connections and failing over cached connections. Don't overrwrite the actual connection id if it has already been set up by a lower level wrapper. --- .../CachingClientConnectionFactory.java | 56 ++++-- .../FailoverClientConnectionFactory.java | 25 +-- .../ip/tcp/TcpOutboundGatewayTests.java | 176 ++++++++++++++++++ .../CachingClientConnectionFactoryTests.java | 42 +++++ 4 files changed, 265 insertions(+), 34 deletions(-) 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 481088cbe6..5889e7d34d 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 @@ -36,8 +36,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact private final SimplePool pool; - private volatile TcpListener listener; - public CachingClientConnectionFactory(AbstractClientConnectionFactory target, int poolSize) { super("", 0); // override single-use to true to force "close" after use @@ -89,7 +87,9 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public TcpConnectionSupport obtainConnection() throws Exception { - return new CachedConnection(this.pool.getItem()); + CachedConnection cachedConnection = new CachedConnection(this.pool.getItem()); + cachedConnection.registerListener(this.getListener()); + return cachedConnection; } private class CachedConnection extends TcpConnectionInterceptorSupport { @@ -134,11 +134,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact return this.getConnectionId(); } - @Override - public TcpListener getListener() { - return CachingClientConnectionFactory.this.listener; - } - /** * 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 @@ -147,10 +142,13 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact */ @Override public boolean onMessage(Message message) { - CachingClientConnectionFactory.this.listener.onMessage(MessageBuilder.fromMessage(message) - .setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId()) - .setHeader(IpHeaders.ACTUAL_CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID)) - .build()); + 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)); + } + this.getListener().onMessage(messageBuilder.build()); close(); // return to pool after response is received return true; // true so the single-use connection doesn't close itself } @@ -273,11 +271,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact return targetConnectionFactory.getPort(); } - @Override - public TcpListener getListener() { - return targetConnectionFactory.getListener(); - } - @Override public TcpSender getSender() { return targetConnectionFactory.getSender(); @@ -298,10 +291,35 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact return targetConnectionFactory.getMapper(); } + /** + * Delegate TCP Client Connection factories that are used to receive + * data need a Listener to send the messages to. + * This applies to client factories used for outbound gateways + * or for a pair of collaborating channel adapters. + *

+ * During initialization, if a factory detects it has no listener + * it's listening logic (active thread) is terminated. + *

+ * The listener registered with a factory is provided to each + * connection it creates so it can call the onMessage() method. + *

+ * This code satisfies the first requirement in that this + * listener signals to the factory that it needs to run + * its listening logic. + *

+ * When we wrap actual connections with CachedConnections, + * the connection is given the wrapper as a listener, so it + * can enhance the headers in onMessage(); the wrapper then invokes + * the real listener supplied here, with the modified message. + */ @Override public void registerListener(TcpListener listener) { - this.listener = listener; - targetConnectionFactory.registerListener(listener); + super.registerListener(listener); + targetConnectionFactory.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + throw new UnsupportedOperationException("This should never be called"); + } + }); } @Override 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 aee6a73fb4..b484b54ac9 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 @@ -100,7 +100,9 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac if (connection != null && connection.isOpen()) { return connection; } - return new FailoverTcpConnection(this.factories); + FailoverTcpConnection failoverTcpConnection = new FailoverTcpConnection(this.factories); + failoverTcpConnection.registerListener(this.getListener()); + return failoverTcpConnection; } @Override @@ -279,11 +281,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac return this.delegate.getDeserializerStateKey(); } - @Override - public void registerListener(TcpListener listener) { - this.delegate.registerListener(listener); - } - @Override public void registerSender(TcpSender sender) { this.delegate.registerSender(sender); @@ -334,11 +331,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac this.delegate.setSerializer(serializer); } - @Override - public TcpListener getListener() { - return this.delegate.getListener(); - } - @Override public long incrementAndGetConnectionSequence() { return this.delegate.incrementAndGetConnectionSequence(); @@ -351,10 +343,13 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac * purposes. */ public boolean onMessage(Message message) { - return FailoverClientConnectionFactory.this.getListener().onMessage(MessageBuilder.fromMessage(message) - .setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId()) - .setHeader(IpHeaders.ACTUAL_CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID)) - .build()); + 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()); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index b68e25a69f..b386809392 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -20,6 +20,10 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.io.IOException; import java.io.ObjectInputStream; @@ -45,16 +49,21 @@ import javax.net.ServerSocketFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.Test; +import org.mockito.Mockito; import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; import org.springframework.integration.Message; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.PollableChannel; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; @@ -178,6 +187,7 @@ public class TcpOutboundGatewayTests { assertTrue(replies.remove("Reply" + i)); } done.set(true); + gateway.stop(); } @Test @@ -256,6 +266,7 @@ public class TcpOutboundGatewayTests { assertTrue(replies.remove("Reply" + i)); } done.set(true); + gateway.stop(); } @Test @@ -390,6 +401,171 @@ public class TcpOutboundGatewayTests { assertEquals(lastReceived.get().replace("Test", "Reply"), replies.get(0)); done.set(true); assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size()); + gateway.stop(); + } + + @Test + public void testCachingFailover() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + final CountDownLatch serverLatch = new CountDownLatch(1); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + while (!done.get()) { + Socket socket = server.accept(); + while (!socket.isClosed()) { + try { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + String request = (String) ois.readObject(); + logger.debug("Read " + request); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + oos.writeObject("bar"); + logger.debug("Replied to " + request); + serverLatch.countDown(); + } + catch (IOException e) { + logger.debug("error on write " + e.getClass().getSimpleName()); + socket.close(); + } + } + } + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + + // Failover + AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class); + TcpConnectionSupport mockConn1 = makeMockConnection(); + when(factory1.getConnection()).thenReturn(mockConn1); + doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class)); + + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port); + factory2.setSerializer(new DefaultSerializer()); + factory2.setDeserializer(new DefaultDeserializer()); + factory2.setSoTimeout(10000); + factory2.setSingleUse(false); + + List factories = new ArrayList(); + factories.add(factory1); + factories.add(factory2); + FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); + failoverFactory.start(); + + // Cache + CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2); + cachingFactory.start(); + TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(cachingFactory); + PollableChannel outputChannel = new QueueChannel(); + gateway.setOutputChannel(outputChannel); + gateway.afterPropertiesSet(); + gateway.start(); + + GenericMessage message = new GenericMessage("foo"); + gateway.handleMessage(message); + Message reply = outputChannel.receive(0); + assertNotNull(reply); + assertEquals("bar", reply.getPayload()); + done.set(true); + gateway.stop(); + verify(mockConn1).send(Mockito.any(Message.class)); + } + + @Test + public void testFailoverCached() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + final CountDownLatch serverLatch = new CountDownLatch(1); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + while (!done.get()) { + Socket socket = server.accept(); + while (!socket.isClosed()) { + try { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + String request = (String) ois.readObject(); + logger.debug("Read " + request); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + oos.writeObject("bar"); + logger.debug("Replied to " + request); + serverLatch.countDown(); + } + catch (IOException e) { + logger.debug("error on write " + e.getClass().getSimpleName()); + socket.close(); + } + } + } + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + + // Cache + AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class); + TcpConnectionSupport mockConn1 = makeMockConnection(); + when(factory1.getConnection()).thenReturn(mockConn1); + doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class)); + CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 1); + + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port); + factory2.setSerializer(new DefaultSerializer()); + factory2.setDeserializer(new DefaultDeserializer()); + factory2.setSoTimeout(10000); + factory2.setSingleUse(false); + CachingClientConnectionFactory cachingFactory2 = new CachingClientConnectionFactory(factory2, 1); + + // Failover + List factories = new ArrayList(); + factories.add(cachingFactory1); + factories.add(cachingFactory2); + FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); + failoverFactory.start(); + + TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(failoverFactory); + PollableChannel outputChannel = new QueueChannel(); + gateway.setOutputChannel(outputChannel); + gateway.afterPropertiesSet(); + gateway.start(); + + GenericMessage message = new GenericMessage("foo"); + gateway.handleMessage(message); + Message reply = outputChannel.receive(0); + assertNotNull(reply); + assertEquals("bar", reply.getPayload()); + done.set(true); + gateway.stop(); + verify(mockConn1).send(Mockito.any(Message.class)); + } + + public TcpConnectionSupport makeMockConnection() { + TcpConnectionSupport connection = mock(TcpConnectionSupport.class); + when(connection.isOpen()).thenReturn(true); + return connection; } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index cc4ce38517..01673f9b04 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -25,6 +25,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Executors; @@ -33,6 +34,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -350,4 +352,44 @@ public class CachingClientConnectionFactoryTests { okToRun.set(false); } + + @Test + public void testCachedFailover() throws Exception { + // Failover + AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class); + AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class); + List factories = new ArrayList(); + factories.add(factory1); + factories.add(factory2); + TcpConnectionSupport mockConn1 = makeMockConnection(); + TcpConnectionSupport mockConn2 = makeMockConnection(); + when(factory1.getConnection()).thenReturn(mockConn1); + when(factory2.getConnection()).thenReturn(mockConn2); + when(factory1.isActive()).thenReturn(true); + when(factory2.isActive()).thenReturn(true); + doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class)); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(mockConn2).send(Mockito.any(Message.class)); + FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); + failoverFactory.start(); + + // Cache + CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2); + cachingFactory.start(); + TcpConnection conn1 = cachingFactory.getConnection(); + GenericMessage message = new GenericMessage("foo"); + conn1 = cachingFactory.getConnection(); + conn1.send(message); + Mockito.verify(mockConn2).send(message); + } + + public TcpConnectionSupport makeMockConnection() { + TcpConnectionSupport connection = mock(TcpConnectionSupport.class); + when(connection.isOpen()).thenReturn(true); + return connection; + } + }