GH-149: Support multiplexed consumers

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/149

When a consumer is multiplexed, configure the container to listen to multiple queues.
Not currently supported for the polled consumer.

When using a DLQ, determine the routing key from the queue in the failed message (unless
an explicit DLQ name has been provisioned - in which case, the same DLQ will be used
for all queues.

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -16,7 +16,9 @@
 
 package org.springframework.cloud.stream.binder.rabbit.provisioning;
 
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.commons.logging.Log;
@@ -129,6 +131,23 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 	@Override
 	public ConsumerDestination provisionConsumerDestination(String name, String group,
 			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
+		if (!properties.isMultiplex()) {
+			return doProvisionConsumerDestination(name, group, properties);
+		}
+		else {
+			String[] destinations = StringUtils.commaDelimitedListToStringArray(name);
+			List<String> queues = new ArrayList<>();
+			for (String destination : destinations) {
+				ConsumerDestination dest = doProvisionConsumerDestination(destination.trim(), group, properties);
+				queues.add(dest.getName());
+			}
+			return new RabbitConsumerDestination(
+					StringUtils.arrayToCommaDelimitedString(queues.toArray(new String[queues.size()])), null);
+		}
+	}
+
+	private ConsumerDestination doProvisionConsumerDestination(String name, String group,
+			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
 		boolean anonymous = !StringUtils.hasText(group);
 		String  baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
 					: properties.getExtension().isQueueNameGroupOnly() ? group : groupedName(name, group);
@@ -171,7 +190,7 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 			autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName,
 					properties.getExtension());
 		}
-		return new RabbitConsumerDestination(queue, binding);
+		return new RabbitConsumerDestination(queue.getName(), binding);
 	}
 
 	/**
@@ -476,13 +495,23 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 		addToAutoDeclareContext(rootName + ".binding", binding);
 	}
 
-	public void cleanAutoDeclareContext(String name) {
+	public void cleanAutoDeclareContext(ConsumerDestination destination,
+			ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
 		synchronized (this.autoDeclareContext) {
-			removeSingleton(name + ".binding");
-			removeSingleton(name);
-			String dlq = name + ".dlq";
-			removeSingleton(dlq + ".binding");
-			removeSingleton(dlq);
+			String[] names = new String[] { destination.getName() };
+			if (consumerProperties.isMultiplex()) {
+				names = StringUtils.commaDelimitedListToStringArray(destination.getName());
+			}
+			for (int i = 0; i < names.length; i++) {
+				names[i] = names[i].trim();
+			}
+			for (String name : names) {
+				removeSingleton(name + ".binding");
+				removeSingleton(name);
+				String dlq = name + ".dlq";
+				removeSingleton(dlq + ".binding");
+				removeSingleton(dlq);
+			}
 		}
 	}
 
@@ -533,10 +562,10 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 
 	private static final class RabbitConsumerDestination implements ConsumerDestination {
 
-		private final Queue queue;
+		private final String queue;
 		private final Binding binding;
 
-		RabbitConsumerDestination(Queue queue, Binding binding) {
+		RabbitConsumerDestination(String queue, Binding binding) {
 			Assert.notNull(queue, "queue must not be null");
 			this.queue = queue;
 			this.binding = binding;
@@ -552,7 +581,7 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 
 		@Override
 		public String getName() {
-			return this.queue.getName();
+			return this.queue;
 		}
 	}
 

spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc

@@ -63,6 +63,8 @@ You can exclude the class by using the `@SpringBootApplication` annotation.
 
 Starting with version 2.0, the `RabbitMessageChannelBinder` sets the `RabbitTemplate.userPublisherConnection` property to `true` so that the non-transactional producers avoid deadlocks on consumers, which can happen if cached connections are blocked because of a https://www.rabbitmq.com/memory.html[memory alarm] on the broker.
 
+NOTE: Currently, a `multiplex` consumer (a single consumer listening to multiple queues) is only supported for message-driven conssumers; polled consumers can only retrieve messages from a single queue.
+
 == Configuration Options
 
 This section contains settings specific to the RabbitMQ Binder and bound channels.

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -361,7 +361,17 @@ public class RabbitMessageChannelBinder
 		listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
 		listenerContainer.setTxSize(properties.getExtension().getTxSize());
 		listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
-		listenerContainer.setQueueNames(destination);
+		String[] queues;
+		if (properties.isMultiplex()) {
+			queues = StringUtils.commaDelimitedListToStringArray(destination);
+		}
+		else {
+			queues = new String[] { destination };
+		}
+		for (int i = 0; i < queues.length; i++) {
+			queues[i] = queues[i].trim();
+		}
+		listenerContainer.setQueueNames(queues);
 		listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
 		listenerContainer.setMessagePropertiesConverter(
 				RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
@@ -403,6 +413,8 @@ public class RabbitMessageChannelBinder
 	@Override
 	protected PolledConsumerResources createPolledConsumerResources(String name, String group, ConsumerDestination destination,
 			ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
+		Assert.isTrue(!consumerProperties.isMultiplex(),
+				"The Spring Integration polled MessageSource does not currently support muiltiple queues");
 		AmqpMessageSource source = new AmqpMessageSource(this.connectionFactory, destination.getName());
 		source.setRawMessageHeader(true);
 		return new PolledConsumerResources(source,
@@ -568,7 +580,7 @@ public class RabbitMessageChannelBinder
 	@Override
 	protected void afterUnbindConsumer(ConsumerDestination consumerDestination, String group,
 			ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
-		provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName());
+		provisioningProvider.cleanAutoDeclareContext(consumerDestination, consumerProperties);
 	}
 
 	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) {

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -49,6 +49,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitManagementTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.AsyncConsumerStartedEvent;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.support.AmqpHeaders;
@@ -641,8 +642,12 @@ public class RabbitBinderTests extends
 			}
 
 		});
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqtest", "default", moduleInputChannel,
-				consumerProperties);
+		consumerProperties.setMultiplex(true);
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqtest,dlqtest2", "default",
+				moduleInputChannel, consumerProperties);
+		AbstractMessageListenerContainer container = TestUtils.getPropertyValue(consumerBinding,
+				"lifecycle.messageListenerContainer", AbstractMessageListenerContainer.class);
+		assertThat(container.getQueueNames().length).isEqualTo(2);
 
 		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
 		template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo");
@@ -658,6 +663,19 @@ public class RabbitBinderTests extends
 		}
 		assertThat(n).isLessThan(100);
 
+		template.convertAndSend("", TEST_PREFIX + "dlqtest2.default", "bar");
+
+		n = 0;
+		while (n++ < 100) {
+			Object deadLetter = template.receiveAndConvert(TEST_PREFIX + "dlqtest2.default.dlq");
+			if (deadLetter != null) {
+				assertThat(deadLetter).isEqualTo("bar");
+				break;
+			}
+			Thread.sleep(100);
+		}
+		assertThat(n).isLessThan(100);
+
 		consumerBinding.unbind();
 
 		ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.provisioningProvider.autoDeclareContext",
@@ -1009,8 +1027,9 @@ public class RabbitBinderTests extends
 			}
 
 		});
-		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.dlqpubtest", "foo", moduleInputChannel,
-				consumerProperties);
+		consumerProperties.setMultiplex(true);
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.dlqpubtest,foo.dlqpubtest2", "foo",
+				moduleInputChannel, consumerProperties);
 
 		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
 		template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest.foo", "foo");
@@ -1027,6 +1046,20 @@ public class RabbitBinderTests extends
 		}
 		assertThat(n).isLessThan(100);
 
+		template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest2.foo", "bar");
+
+		n = 0;
+		while (n++ < 100) {
+			org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "foo.dlqpubtest2.foo.dlq");
+			if (deadLetter != null) {
+				assertThat(new String(deadLetter.getBody())).isEqualTo("bar");
+				assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace"));
+				break;
+			}
+			Thread.sleep(100);
+		}
+		assertThat(n).isLessThan(100);
+
 		consumerBinding.unbind();
 	}
 

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java

@@ -36,6 +36,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.util.StringUtils;
 
 /**
  * Test support class for {@link RabbitMessageChannelBinder}.
@@ -92,15 +93,31 @@ public class RabbitTestBinder extends
 
 	private void captureConsumerResources(String name, String group,
 			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
+		String[] names = null;
 		if (group != null) {
 			if (properties.getExtension().isQueueNameGroupOnly()) {
 				this.queues.add(properties.getExtension().getPrefix() + group);
 			}
 			else {
-				this.queues.add(properties.getExtension().getPrefix() + name + ("." + group));
+				if (properties.isMultiplex()) {
+					names = StringUtils.commaDelimitedListToStringArray(name);
+					for (String nayme : names) {
+						this.queues.add(properties.getExtension().getPrefix() + nayme.trim() + "." + group);
+					}
+				}
+				else {
+					this.queues.add(properties.getExtension().getPrefix() + name + "." + group);
+				}
 			}
 		}
-		this.exchanges.add(properties.getExtension().getPrefix() + name);
+		if (names != null) {
+			for (String nayme : names) {
+				this.exchanges.add(properties.getExtension().getPrefix() + nayme.trim());
+			}
+		}
+		else {
+			this.exchanges.add(properties.getExtension().getPrefix() + name);
+		}
 		this.prefixes.add(properties.getExtension().getPrefix());
 		deadLetters(properties.getExtension());
 	}