From fee8a97532b91c4194df4e3a14e3754b2ce4c899 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 17 Jun 2019 10:19:12 -0400 Subject: [PATCH] GH-1026: Fix Delay with CacheMode.CONNECTION Fixes https://github.com/spring-projects/spring-amqp/issues/1026 When using a `channelCheckoutTimeout` with `CacheModeConnection`, we incorrectly spin waiting for a connection until the timeout expires. We should only wait for a connection if the limit is exceeded. **cherry-pick to all supported** (cherry picked from commit dde7a37350e21a5bbcce26122fbc49b3dd72238b) # Conflicts: # spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java # spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java # Conflicts: # spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java --- .../connection/CachingConnectionFactory.java | 20 +++++++++-------- .../CachingConnectionFactoryTests.java | 22 +++++++++++++++++++ 2 files changed, 33 insertions(+), 9 deletions(-) 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)); + } + }