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 9d34f21731..d98c2562ff 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 @@ -15,7 +15,6 @@ */ package org.springframework.integration.ip.tcp.connection; -import java.io.IOException; import java.util.Iterator; import java.util.List; import java.util.UUID; @@ -177,7 +176,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac * This allows for the condition where the current connection is closed, * the current factory can serve up a new connection, but all other * factories are down. - * @throws Exception + * @throws Exception if an exception occurs */ private synchronized void findAConnection() throws Exception { boolean success = false; @@ -191,11 +190,19 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac try { nextFactory = this.factoryIterator.next(); this.delegate = nextFactory.getConnection(); + if (logger.isDebugEnabled()) { + logger.debug("Got " + this.delegate.getConnectionId() + " from " + nextFactory); + } this.delegate.registerListener(this); this.currentFactory = nextFactory; success = this.delegate.isOpen(); } - catch (IOException e) { + catch (Exception e) { + if (logger.isDebugEnabled()) { + logger.debug(nextFactory + " failed with " + + e.toString() + + ", trying another"); + } if (!this.factoryIterator.hasNext()) { if (retried && lastFactoryToTry == null || lastFactoryToTry == nextFactory) { /* @@ -240,7 +247,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac this.delegate.send(message); success = true; } - catch (IOException e) { + catch (Exception e) { if (retried && lastFactoryTried == lastFactoryToTry) { logger.error("All connection factories exhausted", e); this.open = false; 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 68f56a3005..f2a9bf5291 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 @@ -111,7 +111,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling throw e; } if (logger.isDebugEnabled()) { - logger.debug("Message sent " + message); + logger.debug(getConnectionId() + " Message sent " + message); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index 71b2172a47..8d3aab6d59 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -156,7 +156,7 @@ public class TcpNioConnection extends TcpConnectionSupport { throw e; } if (logger.isDebugEnabled()) { - logger.debug("Message sent " + message); + logger.debug(getConnectionId() + " Message sent " + message); } } } 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 d4be32edb9..6c791b880c 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 @@ -75,6 +75,7 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.SimplePool; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; @@ -567,6 +568,170 @@ public class CachingClientConnectionFactoryTests { Mockito.verify(mockConn2).send(message); } + @Test + public void testCachedFailoverRealClose() throws Exception { + int port1 = SocketUtils.findAvailableTcpPort(); + TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1); + server1.setBeanName("server1"); + final CountDownLatch latch1 = new CountDownLatch(3); + server1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch1.countDown(); + return false; + } + }); + server1.start(); + TestingUtilities.waitListening(server1, 10000L); + int port2 = SocketUtils.findAvailableTcpPort(); + TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2); + server1.setBeanName("server2"); + final CountDownLatch latch2 = new CountDownLatch(2); + server2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch2.countDown(); + return false; + } + }); + server2.start(); + TestingUtilities.waitListening(server2, 10000L); + // Failover + AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("localhost", port1); + factory1.setBeanName("client1"); + factory1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2); + factory2.setBeanName("client2"); + factory2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + List factories = new ArrayList(); + factories.add(factory1); + factories.add(factory2); + FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); + + // Cache + CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2); + cachingFactory.start(); + TcpConnection conn1 = cachingFactory.getConnection(); + GenericMessage message = new GenericMessage("foo"); + conn1.send(message); + conn1.close(); + TcpConnection conn2 = cachingFactory.getConnection(); + assertSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(), + ((TcpConnectionInterceptorSupport) conn2).getTheConnection()); + conn2.send(message); + conn1 = cachingFactory.getConnection(); + assertNotSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(), + ((TcpConnectionInterceptorSupport) conn2).getTheConnection()); + conn1.send(message); + conn1.close(); + conn2.close(); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + server1.stop(); + TestingUtilities.waitStopListening(server1, 10000L); + TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(factory1, 0); + conn1 = cachingFactory.getConnection(); + conn2 = cachingFactory.getConnection(); + conn1.send(message); + conn2.send(message); + conn1.close(); + conn2.close(); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + SimplePool pool = TestUtils.getPropertyValue(cachingFactory, "pool", SimplePool.class); + assertEquals(2, pool.getIdleCount()); + server2.stop(); + } + + @Test + public void testCachedFailoverRealBadHost() throws Exception { + int port1 = SocketUtils.findAvailableTcpPort(); + TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1); + server1.setBeanName("server1"); + final CountDownLatch latch1 = new CountDownLatch(3); + server1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch1.countDown(); + return false; + } + }); + server1.start(); + TestingUtilities.waitListening(server1, 10000L); + int port2 = SocketUtils.findAvailableTcpPort(); + TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2); + server1.setBeanName("server2"); + final CountDownLatch latch2 = new CountDownLatch(2); + server2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch2.countDown(); + return false; + } + }); + server2.start(); + TestingUtilities.waitListening(server2, 10000L); + // Failover + AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("junkjunk", port1); + factory1.setBeanName("client1"); + factory1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2); + factory2.setBeanName("client2"); + factory2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + List factories = new ArrayList(); + factories.add(factory1); + factories.add(factory2); + FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); + + // Cache + CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2); + cachingFactory.start(); + TcpConnection conn1 = cachingFactory.getConnection(); + GenericMessage message = new GenericMessage("foo"); + conn1.send(message); + conn1.close(); + TcpConnection conn2 = cachingFactory.getConnection(); + assertSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(), + ((TcpConnectionInterceptorSupport) conn2).getTheConnection()); + conn2.send(message); + conn1 = cachingFactory.getConnection(); + assertNotSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(), + ((TcpConnectionInterceptorSupport) conn2).getTheConnection()); + conn1.send(message); + conn1.close(); + conn2.close(); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertEquals(3, latch1.getCount()); + server1.stop(); + server2.stop(); + } + @Test //INT-3650 public void testRealConnection() throws Exception { int port = SocketUtils.findAvailableTcpPort(); 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 9cae76387a..4d95d1ff48 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 @@ -16,7 +16,10 @@ package org.springframework.integration.ip.tcp.connection; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.doAnswer; @@ -30,8 +33,10 @@ import java.net.Socket; import java.nio.channels.SocketChannel; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -55,6 +60,7 @@ import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.test.rule.Log4jLevelAdjuster; import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.SimplePool; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -71,7 +77,7 @@ public class FailoverClientConnectionFactoryTests { @Rule public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, - "org.springframework.integration.ip.tcp"); + "org.springframework.integration.ip.tcp", "org.springframework.integration.util.SimplePool"); @Test public void testFailoverGood() throws Exception { @@ -303,6 +309,190 @@ public class FailoverClientConnectionFactoryTests { testRealGuts(client1, client2, server1, server2); } + @Test + public void testFailoverCachedRealClose() throws Exception { + int port1 = SocketUtils.findAvailableServerSocket(); + TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1); + server1.setBeanName("server1"); + final CountDownLatch latch1 = new CountDownLatch(3); + server1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch1.countDown(); + return false; + } + }); + server1.start(); + TestingUtilities.waitListening(server1, 10000L); + int port2 = SocketUtils.findAvailableServerSocket(); + TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2); + server2.setBeanName("server2"); + final CountDownLatch latch2 = new CountDownLatch(2); + server2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch2.countDown(); + return false; + } + }); + server2.start(); + TestingUtilities.waitListening(server2, 10000L); + AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("localhost", port1); + factory1.setBeanName("client1"); + factory1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2); + factory2.setBeanName("client2"); + factory2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + // Cache + CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 2); + cachingFactory1.setBeanName("cache1"); + CachingClientConnectionFactory cachingFactory2 = new CachingClientConnectionFactory(factory2, 2); + cachingFactory2.setBeanName("cache2"); + + // Failover + List factories = new ArrayList(); + factories.add(cachingFactory1); + factories.add(cachingFactory2); + FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); + + failoverFactory.start(); + TcpConnection conn1 = failoverFactory.getConnection(); + conn1.send(new GenericMessage("foo1")); + conn1.close(); + TcpConnection conn2 = failoverFactory.getConnection(); + assertSame( + (TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection(), + (TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection()); + conn2.send(new GenericMessage("foo2")); + conn1 = failoverFactory.getConnection(); + assertNotSame( + (TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection(), + (TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection()); + conn1.send(new GenericMessage("foo3")); + conn1.close(); + conn2.close(); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + server1.stop(); + TestingUtilities.waitStopListening(server1, 10000L); + TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(factory1, 0); + conn1 = failoverFactory.getConnection(); + conn2 = failoverFactory.getConnection(); + conn1.send(new GenericMessage("foo4")); + conn2.send(new GenericMessage("foo5")); + conn1.close(); + conn2.close(); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + SimplePool pool = TestUtils.getPropertyValue(cachingFactory2, "pool", SimplePool.class); + assertEquals(2, pool.getIdleCount()); + server2.stop(); + } + + @Test + public void testFailoverCachedRealBadHost() throws Exception { + int port1 = SocketUtils.findAvailableServerSocket(); + TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1); + server1.setBeanName("server1"); + final CountDownLatch latch1 = new CountDownLatch(3); + server1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch1.countDown(); + return false; + } + }); + server1.start(); + TestingUtilities.waitListening(server1, 10000L); + int port2 = SocketUtils.findAvailableServerSocket(); + TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2); + server2.setBeanName("server2"); + final CountDownLatch latch2 = new CountDownLatch(2); + server2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + latch2.countDown(); + return false; + } + }); + server2.start(); + TestingUtilities.waitListening(server2, 10000L); + + AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("junkjunk", port1); + factory1.setBeanName("client1"); + factory1.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2); + factory2.setBeanName("client2"); + factory2.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + return false; + } + }); + + // Cache + CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 2); + cachingFactory1.setBeanName("cache1"); + CachingClientConnectionFactory cachingFactory2 = new CachingClientConnectionFactory(factory2, 2); + cachingFactory2.setBeanName("cache2"); + + // Failover + List factories = new ArrayList(); + factories.add(cachingFactory1); + factories.add(cachingFactory2); + FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); + failoverFactory.start(); + TcpConnection conn1 = failoverFactory.getConnection(); + GenericMessage message = new GenericMessage("foo"); + conn1.send(message); + conn1.close(); + TcpConnection conn2 = failoverFactory.getConnection(); + assertSame( + (TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection(), + (TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection()); + conn2.send(message); + conn1 = failoverFactory.getConnection(); + assertNotSame( + (TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection(), + (TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class)) + .getTheConnection()); + conn1.send(message); + conn1.close(); + conn2.close(); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertEquals(3, latch1.getCount()); + server1.stop(); + server2.stop(); + } + private void testRealGuts(AbstractClientConnectionFactory client1, AbstractClientConnectionFactory client2, AbstractServerConnectionFactory server1, AbstractServerConnectionFactory server2) throws Exception { int port1 = 0;