GH-804: DRTMLC - don't cancel in-use consumer

Fixes #804

The `DirectReplyToMessageListenerContainer` shuts down consumers that haven't
been used for the `idleEventInterval`. However, the superclass simply cancels
the first consumer in the list, which may actually being used in the DRTMLC.

Change the logic to find a consumer that is not actually being used.

**cherry-pick to 2.0.x**

* Polishing - PR Comments

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

@@ -33,7 +33,6 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import org.apache.commons.logging.Log;
@@ -311,14 +310,31 @@ private void adjustConsumers(int newCount) {
 				}
 				List<SimpleConsumer> consumerList = this.consumersByQueue.get(queue);
 				if (consumerList != null && consumerList.size() > newCount) {
-					IntStream.range(newCount, consumerList.size())
-							.mapToObj(i -> consumerList.remove(0))
-							.forEach(this::cancelConsumer);
+					int delta = consumerList.size() - newCount;
+					for (int i = 0; i < delta; i++) {
+						int index = findIdleConsumer();
+						if (index >= 0) {
+							SimpleConsumer consumer = consumerList.remove(index);
+							if (consumer != null) {
+								cancelConsumer(consumer);
+							}
+						}
+					}
 				}
 			}
 		}
 	}
 
+	/**
+	 * When adjusting down, return a consumer that can be canceled. Called while
+	 * synchronized on consumersMonitor.
+	 * @return the consumer index or -1 if non idle.
+	 * @since 2.0.6
+	 */
+	protected int findIdleConsumer() {
+		return 0;
+	}
+
 	private void checkStartState() {
 		if (!this.isRunning()) {
 			try {

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

@@ -147,6 +147,16 @@ protected void processMonitorTask() {
 		}
 	}
 
+	@Override
+	protected int findIdleConsumer() {
+		for (int i = 0; i < this.consumers.size(); i++) {
+			if (!this.inUseConsumerChannels.containsValue(this.consumers.get(i))) {
+				return i;
+			}
+		}
+		return -1;
+	}
+
 	@Override
 	protected void consumerRemoved(SimpleConsumer consumer) {
 		this.inUseConsumerChannels.remove(consumer.getChannel());

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

@@ -551,12 +551,22 @@ public void testReplyToConsumersReduced() throws Exception {
 		CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
 		DirectReplyToMessageListenerContainer container = new DirectReplyToMessageListenerContainer(cf);
 		container.setBeanName("reducing");
-		container.setIdleEventInterval(500);
+		container.setIdleEventInterval(100);
+		CountDownLatch latch = new CountDownLatch(5);
+		container.setApplicationEventPublisher(e -> {
+			if (e instanceof ListenerContainerIdleEvent) {
+				latch.countDown();
+			}
+		});
 		container.afterPropertiesSet();
 		container.start();
-		ChannelHolder channelHolder = container.getChannelHolder();
-		assertTrue(activeConsumerCount(container, 1));
-		container.releaseConsumerFor(channelHolder, false, null);
+		ChannelHolder channelHolder1 = container.getChannelHolder();
+		ChannelHolder channelHolder2 = container.getChannelHolder();
+		assertTrue(activeConsumerCount(container, 2));
+		container.releaseConsumerFor(channelHolder2, false, null);
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertTrue(channelHolder1.getChannel().isOpen());
+		container.releaseConsumerFor(channelHolder1, false, null);
 		assertTrue(activeConsumerCount(container, 0));
 		container.stop();
 		cf.destroy();