diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index f0f3b26c..07cb501e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -598,15 +598,17 @@ public class CachingConnectionFactory extends AbstractConnectionFactory else if (this.cacheMode == CacheMode.CONNECTION) { ChannelCachingConnectionProxy connection = findIdleConnection(); long now = System.currentTimeMillis(); - while (connection == null && System.currentTimeMillis() - now < this.channelCheckoutTimeout) { - if (countOpenConnections() >= this.connectionLimit) { - try { - this.connectionMonitor.wait(this.channelCheckoutTimeout); - connection = findIdleConnection(); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new AmqpException("Interrupted while waiting for a connection", e); + if (connection == null && countOpenConnections() >= this.connectionLimit) { + while (connection == null && System.currentTimeMillis() - now < this.channelCheckoutTimeout) { + if (countOpenConnections() >= this.connectionLimit) { + try { + this.connectionMonitor.wait(this.channelCheckoutTimeout); + connection = findIdleConnection(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AmqpException("Interrupted while waiting for a connection", e); + } } } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java index 36f3050b..22766b97 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java @@ -16,14 +16,17 @@ package org.springframework.amqp.rabbit.connection; +import static org.hamcrest.Matchers.lessThan; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.AdditionalMatchers.aryEq; +import static org.mockito.BDDMockito.given; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyInt; import static org.mockito.Matchers.anyString; @@ -1597,4 +1600,23 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest Thread.sleep(6000); } + @Test + public void testFirstConnectionDoesntWait() throws IOException, TimeoutException { + com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); + com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); + Channel mockChannel = mock(Channel.class); + + given(mockConnectionFactory.newConnection((ExecutorService) isNull(), anyString())).willReturn(mockConnection); + given(mockConnection.createChannel()).willReturn(mockChannel); + given(mockChannel.isOpen()).willReturn(true); + given(mockConnection.isOpen()).willReturn(true); + + CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); + ccf.setCacheMode(CacheMode.CONNECTION); + ccf.setChannelCheckoutTimeout(60000); + long t1 = System.currentTimeMillis(); + ccf.createConnection(); + assertThat(System.currentTimeMillis() - t1, lessThan(30_000L)); + } + }