GH-935: Handle all exceptions in handleDelivery

Fixes https://github.com/spring-projects/spring-amqp/issues/935

- Don't call basicCancel if already canceled
- Catch all `Exception`s

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

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

@@ -843,7 +843,9 @@ public class BlockingQueueConsumer {
 
 	private final class InternalConsumer extends DefaultConsumer {
 
-		private InternalConsumer(Channel channel) {
+		boolean canceled;
+
+		InternalConsumer(Channel channel) {
 			super(channel);
 		}
 
@@ -890,11 +892,12 @@ public class BlockingQueueConsumer {
 						+ BlockingQueueConsumer.this.consumerTags.get(consumerTag)
 						+ "); " + BlockingQueueConsumer.this);
 			}
+			this.canceled = true;
 		}
 
 		@Override
-		public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
-				throws IOException {
+		public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
+				byte[] body) {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Storing delivery for consumerTag: '"
 						+ consumerTag + "' with deliveryTag: '" + envelope.getDeliveryTag() + "' in "
@@ -907,12 +910,13 @@ public class BlockingQueueConsumer {
 						RabbitUtils.setPhysicalCloseRequired(getChannel(), true);
 						// Defensive - should never happen
 						BlockingQueueConsumer.this.queue.clear();
-						getChannel().basicNack(envelope.getDeliveryTag(), true, true);
-						getChannel().basicCancel(consumerTag);
+						if (!this.canceled) {
+							getChannel().basicCancel(consumerTag);
+						}
 						try {
 							getChannel().close();
 						}
-						catch (TimeoutException e) {
+						catch (@SuppressWarnings("unused") TimeoutException e) {
 							// no-op
 						}
 					}
@@ -921,9 +925,12 @@ public class BlockingQueueConsumer {
 					BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body));
 				}
 			}
-			catch (InterruptedException e) {
+			catch (@SuppressWarnings("unused") InterruptedException e) {
 				Thread.currentThread().interrupt();
 			}
+			catch (Exception e) {
+				BlockingQueueConsumer.logger.warn("Unexpected exception during delivery", e);
+			}
 		}
 
 	}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java

@@ -26,6 +26,7 @@ import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyBoolean;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -40,8 +41,10 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.log4j.Level;
 import org.junit.Rule;
@@ -316,8 +319,7 @@ public class BlockingQueueConsumerTests {
 	}
 
 	@Test
-	@SuppressWarnings("unchecked")
-	public void testDrainAndReject() throws IOException {
+	public void testDrainAndReject() throws IOException, TimeoutException {
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
 		Connection connection = mock(Connection.class);
 		ChannelProxy channel = mock(ChannelProxy.class);
@@ -330,9 +332,18 @@ public class BlockingQueueConsumerTests {
 		doReturn(isOpen.get()).when(channel).isOpen();
 		when(channel.queueDeclarePassive(anyString()))
 				.then(invocation -> mock(AMQP.Queue.DeclareOk.class));
-		when(channel.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
-				any(Map.class), any(Consumer.class))).thenReturn("consumerTag");
-
+		AtomicReference<Consumer> theConsumer = new AtomicReference<>();
+		doAnswer(inv -> {
+			Consumer consumer = (Consumer) inv.getArguments()[6];
+			consumer.handleConsumeOk("consumerTag");
+			theConsumer.set(consumer);
+			return "consumerTag";
+		}).when(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
+				any(), any(Consumer.class));
+		doAnswer(inv -> {
+			theConsumer.get().handleCancelOk("consumerTag");
+			return null;
+		}).when(channel).basicCancel("consumerTag");
 		BlockingQueueConsumer blockingQueueConsumer = new BlockingQueueConsumer(connectionFactory,
 				new DefaultMessagePropertiesConverter(), new ActiveObjectCounter<BlockingQueueConsumer>(),
 				AcknowledgeMode.AUTO, true, 2, "test");
@@ -357,9 +368,7 @@ public class BlockingQueueConsumerTests {
 		envelope = new Envelope(3, false, "foo", "bar");
 		consumer.handleDelivery("consumerTag", envelope, props, new byte[0]);
 		assertThat(TestUtils.getPropertyValue(blockingQueueConsumer, "queue", BlockingQueue.class).size(), equalTo(0));
-		verify(channel).basicNack(3, true, true);
-		verify(channel, times(2)).basicCancel("consumerTag");
+		verify(channel, times(1)).basicCancel("consumerTag");
 	}
 
-
 }