Close auto-recovering connection

When a connection is auto-recovered, the `RabbitAdmin` is not
invoked to re-declare auto-delete queues because the connection
listeners are not invoked.

Close an auto-recovered connection before it is recovered.

Tested with a stand-alone spring-cloud-bus application.

**Cherry-pick to 2.0.x, 1.7.x**

# Conflicts:
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java

Author: garyrussell

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

@@ -374,7 +374,27 @@ protected final Connection createBareConnection() {
 				}
 				rabbitConnection = this.rabbitConnectionFactory.newConnection(this.executorService, connectionName);
 			}
-			Connection connection = new SimpleConnection(rabbitConnection, this.closeTimeout);
+			final Connection connection = new SimpleConnection(rabbitConnection, this.closeTimeout);
+			if (rabbitConnection instanceof AutorecoveringConnection) {
+				((AutorecoveringConnection) rabbitConnection).addRecoveryListener(new RecoveryListener() {
+
+					@Override
+					public void handleRecoveryStarted(Recoverable recoverable) {
+						handleRecovery(recoverable);
+					}
+
+					@Override
+					public void handleRecovery(Recoverable recoverable) {
+						try {
+							connection.close();
+						}
+						catch (Exception e) {
+							AbstractConnectionFactory.this.logger.error("Failed to close auto-recover connection", e);
+						}
+					}
+
+				});
+			}
 			if (this.logger.isInfoEnabled()) {
 				this.logger.info("Created new connection: " + connectionName + "/" + connection);
 			}

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

@@ -25,14 +25,12 @@
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
-import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 
-import java.io.IOException;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.util.ArrayList;
@@ -58,7 +56,6 @@
 import org.junit.rules.ExpectedException;
 
 import org.springframework.amqp.AmqpAuthenticationException;
-import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.AmqpResourceNotAvailableException;
 import org.springframework.amqp.AmqpTimeoutException;
@@ -76,12 +73,8 @@
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
-import com.rabbitmq.client.Recoverable;
-import com.rabbitmq.client.RecoveryListener;
 import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
-import com.rabbitmq.client.impl.recovery.AutorecoveringChannel;
-import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 
 /**
  * @author Dave Syer
@@ -171,13 +164,15 @@ public void testCachedConnectionsChannelLimit() throws Exception {
 			channels.add(connections.get(0).createChannel(false));
 			fail("Exception expected");
 		}
-		catch (AmqpTimeoutException e) { }
+		catch (AmqpTimeoutException e) {
+		}
 		channels.add(connections.get(1).createChannel(false));
 		try {
 			channels.add(connections.get(1).createChannel(false));
 			fail("Exception expected");
 		}
-		catch (AmqpTimeoutException e) { }
+		catch (AmqpTimeoutException e) {
+		}
 		channels.get(0).close();
 		channels.get(1).close();
 		channels.add(connections.get(0).createChannel(false));
@@ -324,6 +319,7 @@ public void testMixTransactionalAndNonTransactional() throws Exception {
 		exception.expect(AmqpIOException.class);
 
 		template2.execute(new ChannelCallback<Void>() {
+
 			@Override
 			public Void doInRabbit(Channel channel) throws Exception {
 				// Should be an exception because the channel is not transactional
@@ -346,9 +342,11 @@ public void testHardErrorAndReconnectNoAuto() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
 		try {
 			template.execute(new ChannelCallback<Object>() {
+
 				@Override
 				public Object doInRabbit(Channel channel) throws Exception {
 					channel.getConnection().addShutdownListener(new ShutdownListener() {
+
 						@Override
 						public void shutdownCompleted(ShutdownSignalException cause) {
 							logger.info("Error", cause);
@@ -377,108 +375,6 @@ public void shutdownCompleted(ShutdownSignalException cause) {
 		assertEquals(null, result);
 	}
 
-	@Test
-	public void testHardErrorAndReconnectAuto() throws Exception {
-		this.connectionFactory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(true);
-		Log cfLogger = spyOnLogger(this.connectionFactory);
-		willReturn(true).given(cfLogger).isDebugEnabled();
-		RabbitTemplate template = new RabbitTemplate(connectionFactory);
-		RabbitAdmin admin = new RabbitAdmin(connectionFactory);
-		Queue queue = new Queue(CF_INTEGRATION_TEST_QUEUE);
-		admin.declareQueue(queue);
-		final String route = queue.getName();
-
-		final CountDownLatch latch = new CountDownLatch(1);
-		final CountDownLatch recoveryLatch = new CountDownLatch(1);
-		final RecoveryListener channelRecoveryListener = new RecoveryListener() {
-
-			@Override
-			public void handleRecoveryStarted(Recoverable recoverable) {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Channel recovery started: " + asString(recoverable));
-				}
-			}
-
-			@Override
-			public void handleRecovery(Recoverable recoverable) {
-				try {
-					((Channel) recoverable).basicCancel("testHardErrorAndReconnect");
-				}
-				catch (IOException e) {
-				}
-				if (logger.isDebugEnabled()) {
-					logger.debug("Channel recovery complete: " + asString(recoverable));
-				}
-			}
-
-			private String asString(Recoverable recoverable) {
-				// TODO: https://github.com/rabbitmq/rabbitmq-java-client/issues/217
-				return ((AutorecoveringChannel) recoverable).getDelegate().toString();
-			}
-
-		};
-		final RecoveryListener connectionRecoveryListener = new RecoveryListener() {
-
-			@Override
-			public void handleRecoveryStarted(Recoverable recoverable) {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Connection recovery started: " + recoverable);
-				}
-			}
-
-			@Override
-			public void handleRecovery(Recoverable recoverable) {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Connection recovery complete: " + recoverable);
-				}
-				recoveryLatch.countDown();
-			}
-
-		};
-		Object connection = ((ConnectionProxy) this.connectionFactory.createConnection()).getTargetConnection();
-		connection = TestUtils.getPropertyValue(connection, "delegate");
-		if (connection instanceof AutorecoveringConnection) {
-			((AutorecoveringConnection) connection).addRecoveryListener(connectionRecoveryListener);
-		}
-		try {
-			template.execute(channel -> {
-				channel.getConnection().addShutdownListener(cause -> {
-					logger.info("Error", cause);
-					latch.countDown();
-					// This will be thrown on the Connection thread just before it dies, so basically ignored
-					throw new RuntimeException(cause);
-				});
-				Channel targetChannel = ((ChannelProxy) channel).getTargetChannel();
-				if (targetChannel instanceof AutorecoveringChannel) {
-					((AutorecoveringChannel) targetChannel).addRecoveryListener(channelRecoveryListener);
-				}
-				else {
-					fail("Expected an AutorecoveringChannel");
-				}
-				String tag = channel.basicConsume(route, false, "testHardErrorAndReconnect",
-						new DefaultConsumer(channel));
-				// Consume twice with the same tag is a hard error (connection will be reset)
-				String result = channel.basicConsume(route, false, tag, new DefaultConsumer(channel));
-				fail("Expected IOException, got: " + result);
-				return null;
-			});
-			fail("Expected AmqpIOException");
-		}
-		catch (AmqpException e) {
-			// expected
-		}
-		assertTrue(recoveryLatch.await(10, TimeUnit.SECONDS));
-		if (logger.isDebugEnabled()) {
-			logger.debug("Resuming test after recovery complete");
-		}
-		template.convertAndSend(route, "message");
-		assertTrue(latch.await(10, TimeUnit.SECONDS));
-		String result = (String) template.receiveAndConvert(route);
-		assertEquals("message", result);
-		result = (String) template.receiveAndConvert(route);
-		assertEquals(null, result);
-	}
-
 	@Test
 	public void testConnectionCloseLog() {
 		Log logger = spy(TestUtils.getPropertyValue(this.connectionFactory, "logger", Log.class));
@@ -539,7 +435,8 @@ public void hangOnClose() throws Exception {
 								socket.close();
 								proxy.close();
 							}
-							catch (Exception ee) { }
+							catch (Exception ee) {
+							}
 						}
 					}
 				});
@@ -555,7 +452,8 @@ public void hangOnClose() throws Exception {
 							socket.close();
 							proxy.close();
 						}
-						catch (Exception ee) { }
+						catch (Exception ee) {
+						}
 					}
 				}
 				socket.close();