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 124421000a..a117968c33 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 @@ -52,6 +52,8 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac private boolean closeOnRefresh; + private boolean failBack = true; + private volatile long creationTime; /** @@ -88,6 +90,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac Assert.isTrue(!this.cachingDelegates, "'refreshSharedInterval' cannot be changed when using 'CachingClientConnectionFactory` delegates"); this.refreshSharedInterval = refreshSharedInterval; + this.failBack = refreshSharedInterval != Long.MAX_VALUE; } /** @@ -154,7 +157,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac protected TcpConnectionSupport obtainConnection() throws Exception { FailoverTcpConnection sharedConnection = (FailoverTcpConnection) getTheConnection(); boolean shared = !isSingleUse() && !this.cachingDelegates; - boolean refreshShared = shared + boolean refreshShared = this.failBack && shared && sharedConnection != null && System.currentTimeMillis() > this.creationTime + this.refreshSharedInterval; if (sharedConnection != null && sharedConnection.isOpen() && !refreshShared) { 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 fe06e51ea3..4375656853 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 @@ -22,6 +22,7 @@ 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.Matchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.doThrow; @@ -45,8 +46,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; -import org.apache.log4j.Level; -import org.junit.Rule; import org.junit.Test; import org.mockito.InOrder; import org.mockito.Mockito; @@ -63,7 +62,6 @@ import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.TcpInboundGateway; import org.springframework.integration.ip.tcp.TcpOutboundGateway; import org.springframework.integration.ip.util.TestingUtilities; -import org.springframework.integration.test.rule.Log4jLevelAdjuster; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.SimplePool; import org.springframework.messaging.Message; @@ -93,10 +91,6 @@ public class FailoverClientConnectionFactoryTests { }; - @Rule - public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, - "org.springframework.integration.ip.tcp", "org.springframework.integration.util.SimplePool"); - @Test public void testFailoverGood() throws Exception { AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class); @@ -126,15 +120,20 @@ public class FailoverClientConnectionFactoryTests { @Test public void testRefreshShared() throws Exception { - testRefreshShared(false); + testRefreshShared(false, 10_000); } @Test public void testRefreshSharedCloseOnRefresh() throws Exception { - testRefreshShared(true); + testRefreshShared(true, 10_000); } - private void testRefreshShared(boolean closeOnRefresh) throws Exception { + @Test + public void testRefreshSharedInfinite() throws Exception { + testRefreshShared(false, Long.MAX_VALUE); + } + + private void testRefreshShared(boolean closeOnRefresh, long interval) throws Exception { AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class); AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class); List factories = new ArrayList(); @@ -159,21 +158,29 @@ public class FailoverClientConnectionFactoryTests { failoverFactory.start(); TcpConnectionSupport connection = failoverFactory.getConnection(); assertNotNull(TestUtils.getPropertyValue(failoverFactory, "theConnection")); - failoverFactory.setRefreshSharedInterval(10_000); - assertSame(failoverFactory.getConnection(), connection); - failoverFactory.setRefreshSharedInterval(-1); - assertNotSame(failoverFactory.getConnection(), connection); - InOrder inOrder = inOrder(factory1, factory2, conn1); + failoverFactory.setRefreshSharedInterval(interval); + InOrder inOrder = inOrder(factory1, factory2, conn1, conn2); inOrder.verify(factory1).getConnection(); inOrder.verify(factory2).getConnection(); + inOrder.verify(conn1).registerListener(any()); + inOrder.verify(conn1).isOpen(); + assertSame(failoverFactory.getConnection(), connection); + inOrder.verifyNoMoreInteractions(); + failoverFactory.setRefreshSharedInterval(-1); + assertNotSame(failoverFactory.getConnection(), connection); inOrder.verify(factory1).getConnection(); inOrder.verify(factory2).getConnection(); if (closeOnRefresh) { + inOrder.verify(conn2).registerListener(any()); + inOrder.verify(conn2).isOpen(); inOrder.verify(conn1).close(); } else { + inOrder.verify(conn1).registerListener(any()); + inOrder.verify(conn1).isOpen(); inOrder.verify(conn1, never()).close(); } + inOrder.verifyNoMoreInteractions(); } @Test(expected = IOException.class)