GH-1113: Clean up cals to basicCancel

Resolves #1113

Use `RabbitUtils` for all `basicCancel` calls, with try/catch for
all exceptions.

* Add channel to log message.

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

Author: garyrussell

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

@@ -81,7 +81,7 @@ public abstract class RabbitUtils {
 	 */
 	public static final int CHANNEL_PROTOCOL_CLASS_ID_20 = 20;
 
-	private static final Log logger = LogFactory.getLog(RabbitUtils.class); // NOSONAR - lower case
+	private static final Log LOGGER = LogFactory.getLog(RabbitUtils.class);
 
 	private static final ThreadLocal<Boolean> physicalCloseRequired = new ThreadLocal<>(); // NOSONAR - lower case
 
@@ -99,7 +99,7 @@ public static void closeConnection(@Nullable Connection connection) {
 				// empty
 			}
 			catch (Exception ex) {
-				logger.debug("Ignoring Connection exception - assuming already closed: " + ex.getMessage(), ex);
+				LOGGER.debug("Ignoring Connection exception - assuming already closed: " + ex.getMessage(), ex);
 			}
 		}
 	}
@@ -118,15 +118,15 @@ public static void closeChannel(@Nullable Channel channel) {
 				// empty
 			}
 			catch (IOException ex) {
-				logger.debug("Could not close RabbitMQ Channel", ex);
+				LOGGER.debug("Could not close RabbitMQ Channel", ex);
 			}
 			catch (ShutdownSignalException sig) {
 				if (!isNormalShutdown(sig)) {
-					logger.debug("Unexpected exception on closing RabbitMQ Channel", sig);
+					LOGGER.debug("Unexpected exception on closing RabbitMQ Channel", sig);
 				}
 			}
 			catch (Exception ex) {
-				logger.debug("Unexpected exception on closing RabbitMQ Channel", ex);
+				LOGGER.debug("Unexpected exception on closing RabbitMQ Channel", ex);
 			}
 		}
 	}
@@ -182,18 +182,18 @@ public static void closeMessageConsumer(Channel channel, Collection<String> cons
 		}
 	}
 
-	private static void cancel(Channel channel, String consumerTag) {
+	public static void cancel(Channel channel, String consumerTag) {
 		try {
 			channel.basicCancel(consumerTag);
 		}
-		catch (IOException e) {
-			if (logger.isDebugEnabled()) {
-				logger.debug("Error performing 'basicCancel'", e);
+		catch (AlreadyClosedException e) {
+			if (LOGGER.isTraceEnabled()) {
+				LOGGER.trace(channel + " is already closed", e);
 			}
 		}
-		catch (AlreadyClosedException e) {
-			if (logger.isTraceEnabled()) {
-				logger.trace(channel + " is already closed");
+		catch (Exception e) {
+			if (LOGGER.isDebugEnabled()) {
+				LOGGER.debug("Error performing 'basicCancel' on " + channel, e);
 			}
 		}
 	}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -1842,14 +1842,7 @@ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProp
 	}
 
 	private void cancelConsumerQuietly(Channel channel, DefaultConsumer consumer) {
-		try {
-			channel.basicCancel(consumer.getConsumerTag());
-		}
-		catch (Exception e) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug("Failed to cancel consumer: " + consumer, e);
-			}
-		}
+		RabbitUtils.cancel(channel, consumer.getConsumerTag());
 	}
 
 	@Nullable

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

@@ -67,7 +67,6 @@
 import org.springframework.util.backoff.BackOffExecution;
 
 import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
@@ -403,20 +402,8 @@ protected void basicCancel() {
 	protected void basicCancel(boolean expected) {
 		this.normalCancel = expected;
 		getConsumerTags().forEach(consumerTag -> {
-			try {
-				if (this.channel.isOpen()) {
-					this.channel.basicCancel(consumerTag);
-				}
-			}
-			catch (IOException | IllegalStateException e) {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Error performing 'basicCancel'", e);
-				}
-			}
-			catch (AlreadyClosedException e) {
-				if (logger.isTraceEnabled()) {
-					logger.trace(this.channel + " is already closed");
-				}
+			if (this.channel.isOpen()) {
+				RabbitUtils.cancel(this.channel, consumerTag);
 			}
 		});
 		this.cancelled.set(true);
@@ -911,7 +898,7 @@ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProp
 						// Defensive - should never happen
 						BlockingQueueConsumer.this.queue.clear();
 						if (!this.canceled) {
-							getChannel().basicCancel(consumerTag);
+							RabbitUtils.cancel(channelToClose, consumerTag);
 						}
 						try {
 							channelToClose.close();

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java

@@ -816,13 +816,15 @@ private void cancelConsumer(SimpleConsumer consumer) {
 			synchronized (consumer) {
 				consumer.setCanceled(true);
 				if (this.messagesPerAck > 1) {
-					consumer.ackIfNecessary(0L);
+					try {
+						consumer.ackIfNecessary(0L);
+					}
+					catch (IOException e) {
+						this.logger.error("Exception while sending delayed ack", e);
+					}
 				}
 			}
-			consumer.getChannel().basicCancel(consumer.getConsumerTag());
-		}
-		catch (IOException e) {
-			this.logger.error("Failed to cancel consumer: " + consumer, e);
+			RabbitUtils.cancel(consumer.getChannel(), consumer.getConsumerTag());
 		}
 		finally {
 			this.consumers.remove(consumer);