AMQP-836: SMLC queuesChanged() via Queue

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

Changing queues with add/remove with `Queue` parameters did not
restart consumers. Changing the queues with queue names worked.

**cherry-pick to 2.0.x**

(cherry picked from commit 2480b17)

Author: garyrussell

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

@@ -41,6 +41,7 @@
 import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
 import org.springframework.amqp.rabbit.connection.ConsumerChannelRegistry;
@@ -336,7 +337,7 @@ public void setMissingQueuesFatal(boolean missingQueuesFatal) {
 	@Override
 	public void setQueueNames(String... queueName) {
 		super.setQueueNames(queueName);
-		this.queuesChanged();
+		queuesChanged();
 	}
 
 	/**
@@ -349,7 +350,7 @@ public void setQueueNames(String... queueName) {
 	@Override
 	public void addQueueNames(String... queueName) {
 		super.addQueueNames(queueName);
-		this.queuesChanged();
+		queuesChanged();
 	}
 
 	/**
@@ -361,7 +362,37 @@ public void addQueueNames(String... queueName) {
 	@Override
 	public boolean removeQueueNames(String... queueName) {
 		if (super.removeQueueNames(queueName)) {
-			this.queuesChanged();
+			queuesChanged();
+			return true;
+		}
+		else {
+			return false;
+		}
+	}
+
+	/**
+	 * Add queue(s) to this container's list of queues. The existing consumers
+	 * will be cancelled after they have processed any pre-fetched messages and
+	 * new consumers will be created. The queue must exist to avoid problems when
+	 * restarting the consumers.
+	 * @param queue The queue to add.
+	 */
+	@Override
+	public void addQueues(Queue... queue) {
+		super.addQueues(queue);
+		queuesChanged();
+	}
+
+	/**
+	 * Remove queues from this container's list of queues. The existing consumers
+	 * will be cancelled after they have processed any pre-fetched messages and
+	 * new consumers will be created. At least one queue must remain.
+	 * @param queue The queue to remove.
+	 */
+	@Override
+	public boolean removeQueues(Queue... queue) {
+		if (super.removeQueues(queue)) {
+			queuesChanged();
 			return true;
 		}
 		else {

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

@@ -165,6 +165,31 @@ public void testChangeQueues() throws Exception {
 		assertNull(template.receiveAndConvert(queue1.getName()));
 	}
 
+	@Test
+	public void testChangeQueues2() throws Exception { // addQueues instead of addQueueNames
+		CountDownLatch latch = new CountDownLatch(30);
+		container = createContainer(new MessageListenerAdapter(new PojoListener(latch)), queue.getName(), queue1.getName());
+		final CountDownLatch consumerLatch = new CountDownLatch(1);
+		this.container.setApplicationEventPublisher(e -> {
+			if (e instanceof AsyncConsumerStoppedEvent) {
+				consumerLatch.countDown();
+			}
+		});
+		for (int i = 0; i < 10; i++) {
+			template.convertAndSend(queue.getName(), i + "foo");
+			template.convertAndSend(queue1.getName(), i + "foo");
+		}
+		container.addQueues(queue1);
+		assertTrue(consumerLatch.await(10, TimeUnit.SECONDS));
+		for (int i = 0; i < 10; i++) {
+			template.convertAndSend(queue.getName(), i + "foo");
+		}
+		boolean waited = latch.await(10, TimeUnit.SECONDS);
+		assertTrue("Timed out waiting for message", waited);
+		assertNull(template.receiveAndConvert(queue.getName()));
+		assertNull(template.receiveAndConvert(queue1.getName()));
+	}
+
 	@Test
 	public void testNoQueues() throws Exception {
 		CountDownLatch latch1 = new CountDownLatch(20);