Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1026: Fix Delay with CacheMode.CONNECTION

Browse Source 
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 dde7a37350)

# 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
This commit is contained in:
Gary Russell
2019-06-17 10:19:12 -04:00
committed by Artem Bilan
parent 34e614a30a
commit fee8a97532
2 changed files with 33 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -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);
}
}
}
}

22
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

@@ -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));
}
}