From 1f2706ca61b7d699ee1176be6b65f2040fe3b1af Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 8 Sep 2015 10:13:23 -0400 Subject: [PATCH] AMQP-533: @RabbitListener SpEL Multiple Queues JIRA: https://jira.spring.io/browse/AMQP-533 --- ...itListenerAnnotationBeanPostProcessor.java | 42 ++++++++++++------- .../EnableRabbitIntegrationTests.java | 37 +++++++++++++++- src/reference/asciidoc/amqp.adoc | 37 ++++++++++++++++ src/reference/asciidoc/whats-new.adoc | 3 ++ 4 files changed, 104 insertions(+), 15 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java index 7f62f432..3ff39ddb 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java +++ b/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 result = new ArrayList(); 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 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) 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) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 8b85b5c7..8303c90d 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/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 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 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 list = new ArrayList(); + list.add(comma3); + list.add(comma4); + return list; + } + @Bean public ConsumerTagStrategy consumerTagStrategy() { return new ConsumerTagStrategy() { diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 85cf09d3..f5f95c4e 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 2c73da13..795c534b 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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 <> for more information.