AMQP-533: @RabbitListener SpEL Multiple Queues

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

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java

@@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.annotation;
 
 import java.lang.reflect.Method;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -352,28 +353,41 @@ public class RabbitListenerAnnotationBeanPostProcessor
 		if (queues.length > 0 && bindings.length > 0) {
 			throw new BeanInitializationException("@RabbitListener can have 'queues' or 'bindings' but not both");
 		}
-		int length = queues.length > 0 ? queues.length : bindings.length;
-		String[] result = new String[length];
+		List<String> result = new ArrayList<String>();
 		if (queues.length > 0) {
 			for (int i = 0; i < queues.length; i++) {
 				Object resolvedValue = resolveExpression(queues[i]);
-				if (resolvedValue instanceof Queue) {
-					result[i] = ((Queue) resolvedValue).getName();
-				}
-				else if (resolvedValue instanceof String) {
-					result[i] = (String) resolvedValue;
-				}
-				else {
-					throw new IllegalArgumentException(String.format(
-							"@RabbitListener can't resolve '%s' as either a String or a Queue",
-							resolvedValue));
-				}
+				resolveAsString(resolvedValue, result);
 			}
 		}
 		else {
 			return registerBeansForDeclaration(rabbitListener);
 		}
-		return result;
+		return result.toArray(new String[result.size()]);
+	}
+
+	@SuppressWarnings("unchecked")
+	private void resolveAsString(Object resolvedValue, List<String> result) {
+		Object resolvedValueToUse = resolvedValue;
+		if (resolvedValue instanceof String[]) {
+			resolvedValueToUse = Arrays.asList((String[]) resolvedValue);
+		}
+		if (resolvedValueToUse instanceof Queue) {
+			result.add(((Queue) resolvedValueToUse).getName());
+		}
+		else if (resolvedValueToUse instanceof String) {
+			result.add((String) resolvedValueToUse);
+		}
+		else if (resolvedValueToUse instanceof Iterable) {
+			for (Object object : (Iterable<Object>) resolvedValueToUse) {
+				resolveAsString(object, result);
+			}
+		}
+		else {
+			throw new IllegalArgumentException(String.format(
+					"@RabbitListener can't resolve '%s' as either a String or a Queue",
+					resolvedValue));
+		}
 	}
 
 	private Object resolveExpression(String value) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.amqp.rabbit.annotation;
 
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.startsWith;
@@ -25,6 +26,9 @@ import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Date;
 import java.util.List;
 import java.util.UUID;
@@ -49,6 +53,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbit.test.BrokerRunning;
 import org.springframework.amqp.rabbit.test.MessageTestUtils;
@@ -83,7 +88,8 @@ public class EnableRabbitIntegrationTests {
 	public static final BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues(
 			"test.simple", "test.header", "test.message", "test.reply", "test.sendTo", "test.sendTo.reply",
 			"test.sendTo.spel", "test.sendTo.reply.spel",
-			"test.invalidPojo");
+			"test.invalidPojo",
+			"test.comma.1", "test.comma.2", "test.comma.3", "test.comma.4", "test,with,commas");
 
 	@Autowired
 	private RabbitTemplate rabbitTemplate;
@@ -138,6 +144,17 @@ public class EnableRabbitIntegrationTests {
 		assertEquals(2, this.context.getBean("testGroup", List.class).size());
 	}
 
+	@Test
+	public void commas() {
+		assertEquals("FOOfoo", rabbitTemplate.convertSendAndReceive("test,with,commas", "foo"));
+		List<?> commaContainers = this.context.getBean("commas", List.class);
+		assertEquals(1, commaContainers.size());
+		SimpleMessageListenerContainer container = (SimpleMessageListenerContainer) commaContainers.get(0);
+		List<String> queueNames = Arrays.asList(container.getQueueNames());
+		assertThat(queueNames,
+				contains("test.comma.1", "test.comma.2", "test,with,commas", "test.comma.3", "test.comma.4"));
+	}
+
 	@Test
 	public void multiListener() {
 		Bar bar = new Bar();
@@ -266,6 +283,14 @@ public class EnableRabbitIntegrationTests {
 			return foo.toUpperCase();
 		}
 
+		@RabbitListener(queues = {"#{'test.comma.1,test.comma.2'.split(',')}",
+								  "test,with,commas",
+								  "#{commaQueues}"},
+						group = "commas")
+		public String multiQueuesConfig(String foo) {
+			return foo.toUpperCase() + foo;
+		}
+
 		@RabbitListener(queues = "test.header", group = "testGroup")
 		public String capitalizeWithHeader(@Payload String content, @Header String prefix) {
 			return prefix + content.toUpperCase();
@@ -327,6 +352,16 @@ public class EnableRabbitIntegrationTests {
 			return UUID.randomUUID().toString();
 		}
 
+		@Bean
+		public Collection<org.springframework.amqp.core.Queue> commaQueues() {
+			org.springframework.amqp.core.Queue comma3 = new org.springframework.amqp.core.Queue("test.comma.3");
+			org.springframework.amqp.core.Queue comma4 = new org.springframework.amqp.core.Queue("test.comma.4");
+			List<org.springframework.amqp.core.Queue> list = new ArrayList<org.springframework.amqp.core.Queue>();
+			list.add(comma3);
+			list.add(comma4);
+			return list;
+		}
+
 		@Bean
 		public ConsumerTagStrategy consumerTagStrategy() {
 			return new ConsumerTagStrategy() {

src/reference/asciidoc/amqp.adoc

@@ -1384,6 +1384,43 @@ public class AppConfig implements RabbitListenerConfigurer {
 }
 ----
 
+[[annotation-multiple-queues]]
+====== Listening to Multiple Queues
+
+When using the `queues` attribute, you can specify that the associated container can listen to multiple queues.
+You can use a `@Header` annotation to make the queue name from which a message was received available to the POJO
+method:
+
+[source, java]
+----
+@Component
+public class MyService {
+
+    @RabbitListener(queues = { "queue1", "queue2" } )
+    public void processOrder(String data, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
+        ...
+    }
+
+}
+----
+
+Starting with _version 1.5_, you can externalize the queue names using property placeholders, and SpEL:
+
+[source, java]
+----
+@Component
+public class MyService {
+
+    @RabbitListener(queues = "#{'${property.with.comma.delimited.queue.names}'.split(',')}" )
+    public void processOrder(String data, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
+        ...
+    }
+
+}
+----
+
+Prior to _version 1.5_, only a single queue could be specified this way; each queue needed a separate property.
+
 [[async-annotation-driven-reply]]
 ====== Reply Management
 

src/reference/asciidoc/whats-new.adoc

@@ -66,6 +66,9 @@ the Broker.
 
 The default reply address (`@SendTo`) for a `@RabbitListener` can now be a SpEL expression.
 
+====== Multiple Queue Names Via Properties
+
+It is now possible to use a combination of SpEL and property placeholders to specify multiple queues for a listener.
 
 See <<async-annotation-driven>> for more information.