GH-1026: Fix Delay with CacheMode.CONNECTION

Fixes #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 dde7a37)

# 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

Author: garyrussell

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

@@ -638,15 +638,17 @@ public final Connection createConnection() throws AmqpException {
 			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);
+							}
 						}
 					}
 				}

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

@@ -16,18 +16,21 @@
 
 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.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
@@ -1654,4 +1657,23 @@ public void testReturnsNormalCloseDeferredClose() throws Exception {
 		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));
+	}
+
 }