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; + } + }