AMQP-498: Adjust Consumers With ChannelTransacted

JIRA: https://jira.spring.io/browse/AMQP-498

In earlier versions, the while loop to drain the queue took an early exit
when the channel was transacted. (This was actually incorrect since message
delivery is not transactional, only acks).

AMQP-388 introduced a new OR condition on the loop: `consumer.hasDelivery()`.

This effectively made the `continuable` boolean in the OR condition irrelevant.
This boolean was true if messages were received and the channel was not transactional.

Another side effect was the `continuable` boolean was incorrectly used to adjust the consumers
if the workload demanded. Instead, that decision should have depended just on whether
messages were received; whether the channel is transacted is irrelevent.

Remove the `continuable` boolean; fix the while loop and use a new boolean `receivedOk` which
indicates whether messages were received and so whether we should consider increasing the
consumers.

Add a test to verify consumers are adjusted for transactional channels.

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

@@ -786,7 +786,8 @@ public class BlockingQueueConsumer {
 				RabbitUtils.commitIfNecessary(channel);
 			}
 
-		} finally {
+		}
+		finally {
 			deliveryTags.clear();
 		}
 

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

@@ -1110,15 +1110,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 					ConsumerChannelRegistry.registerConsumerChannel(consumer.getChannel(), getConnectionFactory());
 				}
 
-				// Always better to stop receiving as soon as possible if
-				// transactional
-				boolean continuable = false;
-				while (isActive(this.consumer) || this.consumer.hasDelivery() || continuable) {
+				while (isActive(this.consumer) || this.consumer.hasDelivery()) {
 					try {
-						// Will come back false when the queue is drained
-						continuable = receiveAndExecute(this.consumer) && !isChannelTransacted();
+						boolean receivedOk = receiveAndExecute(this.consumer); // At least one message received
 						if (SimpleMessageListenerContainer.this.maxConcurrentConsumers != null) {
-							if (continuable) {
+							if (receivedOk) {
 								if (isActive(this.consumer)) {
 									consecutiveIdles = 0;
 									if (consecutiveMessages++ > SimpleMessageListenerContainer.this.consecutiveActiveTrigger) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2014 the original author or authors.
+ * Copyright 2013-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -48,34 +48,49 @@ public class SimpleMessageListenerContainerLongTests {
 	public LongRunningIntegrationTest longTest = new LongRunningIntegrationTest();
 
 	@Rule
-	public BrokerRunning brokerRunning = BrokerRunning.isRunning();
+	public BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues("foo");
 
 	@Test
 	public void testChangeConsumerCount() throws Exception {
+		testChangeConsumerCountGuts(false);
+	}
+
+	@Test
+	public void testChangeConsumerCountTransacted() throws Exception {
+		testChangeConsumerCountGuts(true);
+	}
+
+	private void testChangeConsumerCountGuts(boolean transacted) throws Exception {
 		final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory("localhost");
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(singleConnectionFactory);
-		container.setMessageListener(new MessageListenerAdapter(this));
-		container.setQueueNames("foo");
-		container.setAutoStartup(false);
-		container.setConcurrentConsumers(2);
-		container.afterPropertiesSet();
-		assertEquals(2, ReflectionTestUtils.getField(container, "concurrentConsumers"));
-		container.start();
-		waitForNConsumers(container, 2);
-		container.setConcurrentConsumers(1);
-		waitForNConsumers(container, 1);
-		container.setMaxConcurrentConsumers(3);
-		RabbitTemplate template = new RabbitTemplate(singleConnectionFactory);
-		for (int i = 0; i < 20; i++) {
-			template.convertAndSend("foo", "foo");
+		try {
+			container.setMessageListener(new MessageListenerAdapter(this));
+			container.setQueueNames("foo");
+			container.setAutoStartup(false);
+			container.setConcurrentConsumers(2);
+			container.setChannelTransacted(transacted);
+			container.afterPropertiesSet();
+			assertEquals(2, ReflectionTestUtils.getField(container, "concurrentConsumers"));
+			container.start();
+			waitForNConsumers(container, 2);
+			container.setConcurrentConsumers(1);
+			waitForNConsumers(container, 1);
+			container.setMaxConcurrentConsumers(3);
+			RabbitTemplate template = new RabbitTemplate(singleConnectionFactory);
+			for (int i = 0; i < 20; i++) {
+				template.convertAndSend("foo", "foo");
+			}
+			waitForNConsumers(container, 2);		// increased consumers due to work
+			waitForNConsumers(container, 1, 20000); // should stop the extra consumer after 10 seconds idle
+			container.setConcurrentConsumers(3);
+			waitForNConsumers(container, 3);
+			container.stop();
+			waitForNConsumers(container, 0);
+			singleConnectionFactory.destroy();
+		}
+		finally {
+			container.stop();
 		}
-		waitForNConsumers(container, 2);		// increased consumers due to work
-		waitForNConsumers(container, 1, 20000); // should stop the extra consumer after 10 seconds idle
-		container.setConcurrentConsumers(3);
-		waitForNConsumers(container, 3);
-		container.stop();
-		waitForNConsumers(container, 0);
-		singleConnectionFactory.destroy();
 	}
 
 	@Test