From 30adfae9ec43e7fd17104dc50f2cbcc47b65be6a Mon Sep 17 00:00:00 2001 From: Jasper de Vries Date: Mon, 10 Nov 2014 15:32:45 +0200 Subject: [PATCH] AMQP-439: RabbitListener.queues: add SpEL support JIRA: https://jira.spring.io/browse/AMQP-439 --- .../rabbit/annotation/RabbitListener.java | 4 +- ...itListenerAnnotationBeanPostProcessor.java | 44 ++++- ...tenerAnnotationBeanPostProcessorTests.java | 173 ++++++++++++++++++ .../annotation/queue-annotation.properties | 1 + 4 files changed, 217 insertions(+), 5 deletions(-) create mode 100644 spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/annotation/queue-annotation.properties diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java index 45b04942..44af1bda 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java @@ -97,7 +97,9 @@ public @interface RabbitListener { /** * The queues for this listener. - * @return the queue names to listen to from target + * The entries can be 'queue name', 'property-placeholder keys' or 'expressions'. + * Expression must be resolved to the queue name or {@code Queue} object. + * @return the queue names or expressions (SpEL) to listen to from target * {@link org.springframework.amqp.rabbit.listener.MessageListenerContainer}. */ String[] queues(); 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 fa3fe263..15d7950e 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 @@ -20,6 +20,7 @@ import java.lang.reflect.Method; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint; @@ -34,8 +35,12 @@ import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.beans.factory.config.BeanExpressionContext; +import org.springframework.beans.factory.config.BeanExpressionResolver; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.context.expression.StandardBeanExpressionResolver; import org.springframework.core.Ordered; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; @@ -88,12 +93,16 @@ public class RabbitListenerAnnotationBeanPostProcessor private BeanFactory beanFactory; - private final RabbitHandlerMethodFactoryAdapter messageHandlerMethodFactory = new RabbitHandlerMethodFactoryAdapter(); + private final RabbitHandlerMethodFactoryAdapter messageHandlerMethodFactory = + new RabbitHandlerMethodFactoryAdapter(); private final RabbitListenerEndpointRegistrar registrar = new RabbitListenerEndpointRegistrar(); private final AtomicInteger counter = new AtomicInteger(); + private BeanExpressionResolver resolver = new StandardBeanExpressionResolver(); + + private BeanExpressionContext expressionContext; @Override public int getOrder() { @@ -140,6 +149,10 @@ public class RabbitListenerAnnotationBeanPostProcessor @Override public void setBeanFactory(BeanFactory beanFactory) { this.beanFactory = beanFactory; + if (beanFactory instanceof ConfigurableListableBeanFactory) { + this.resolver = ((ConfigurableListableBeanFactory) beanFactory).getBeanExpressionResolver(); + this.expressionContext = new BeanExpressionContext((ConfigurableListableBeanFactory) beanFactory, null); + } } @@ -157,9 +170,11 @@ public class RabbitListenerAnnotationBeanPostProcessor if (this.registrar.getEndpointRegistry() == null) { if (this.endpointRegistry == null) { - Assert.state(this.beanFactory != null, "BeanFactory must be set to find endpoint registry by bean name"); + Assert.state(this.beanFactory != null, + "BeanFactory must be set to find endpoint registry by bean name"); this.endpointRegistry = this.beanFactory.getBean( - RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, RabbitListenerEndpointRegistry.class); + RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, + RabbitListenerEndpointRegistry.class); } this.registrar.setEndpointRegistry(this.endpointRegistry); } @@ -281,11 +296,32 @@ public class RabbitListenerAnnotationBeanPostProcessor private String[] resolveQueues(String... queues) { String[] result = new String[queues.length]; for (int i = 0; i < queues.length; i++) { - result[i] = resolve(queues[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)); + } } return result; } + private Object resolveExpression(String value) { + String resolvedValue = resolve(value); + + if (!(resolvedValue.startsWith("#{") && value.endsWith("}"))) { + return resolvedValue; + } + + return this.resolver.evaluate(resolvedValue, this.expressionContext); + } + /** * Resolve the specified value if possible. * diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java index 04816def..0aba07d9 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java @@ -18,15 +18,19 @@ package org.springframework.amqp.rabbit.annotation; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; +import java.util.Iterator; +import org.hamcrest.Matchers; import org.junit.Test; +import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.config.MessageListenerTestContainer; import org.springframework.amqp.rabbit.config.RabbitListenerContainerTestFactory; import org.springframework.amqp.rabbit.listener.AbstractRabbitListenerEndpoint; @@ -34,10 +38,13 @@ import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint; import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.PropertySource; +import org.springframework.context.support.PropertySourcesPlaceholderConfigurer; import org.springframework.stereotype.Component; /** @@ -70,6 +77,34 @@ public class RabbitListenerAnnotationBeanPostProcessorTests { assertTrue("Should have been stopped " + container, container.isStopped()); } + @Test + public void simpleMessageListenerWithMixedAnnotations() { + ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( + Config.class, SimpleMessageListenerWithMixedAnnotationsTestBean.class); + + RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class); + assertEquals("One container should have been registered", 1, factory.getListenerContainers().size()); + MessageListenerTestContainer container = factory.getListenerContainers().get(0); + + RabbitListenerEndpoint endpoint = container.getEndpoint(); + assertEquals("Wrong endpoint type", MethodRabbitListenerEndpoint.class, endpoint.getClass()); + MethodRabbitListenerEndpoint methodEndpoint = (MethodRabbitListenerEndpoint) endpoint; + assertNotNull(methodEndpoint.getBean()); + assertNotNull(methodEndpoint.getMethod()); + + Iterator iterator = ((MethodRabbitListenerEndpoint) endpoint).getQueueNames().iterator(); + assertEquals("testQueue", iterator.next()); + assertEquals("secondQueue", iterator.next()); + + SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(); + methodEndpoint.setupListenerContainer(listenerContainer); + assertNotNull(listenerContainer.getMessageListener()); + + assertTrue("Should have been started " + container, container.isStarted()); + context.close(); // Close and stop the listeners + assertTrue("Should have been stopped " + container, container.isStopped()); + } + @Test public void metaAnnotationIsDiscovered() { ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( @@ -83,6 +118,80 @@ public class RabbitListenerAnnotationBeanPostProcessorTests { context.close(); } + @Test + public void multipleQueueNamesTestBean() { + ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( + Config.class, MultipleQueueNamesTestBean.class); + + RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class); + assertEquals("one container should have been registered", 1, factory.getListenerContainers().size()); + RabbitListenerEndpoint endpoint = factory.getListenerContainers().get(0).getEndpoint(); + final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator(); + assertEquals("metaTestQueue", iterator.next()); + assertEquals("testQueue", iterator.next()); + + context.close(); + } + + @Test + public void multipleQueuesTestBean() { + ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( + Config.class, MultipleQueuesTestBean.class); + + RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class); + assertEquals("one container should have been registered", 1, factory.getListenerContainers().size()); + RabbitListenerEndpoint endpoint = factory.getListenerContainers().get(0).getEndpoint(); + final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator(); + assertEquals("testQueue", iterator.next()); + assertEquals("secondQueue", iterator.next()); + + context.close(); + } + + @Test + public void mixedQueuesAndQueueNamesTestBean() { + ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( + Config.class, MixedQueuesAndQueueNamesTestBean.class); + + RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class); + assertEquals("one container should have been registered", 1, factory.getListenerContainers().size()); + RabbitListenerEndpoint endpoint = factory.getListenerContainers().get(0).getEndpoint(); + final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator(); + assertEquals("metaTestQueue", iterator.next()); + assertEquals("testQueue", iterator.next()); + assertEquals("secondQueue", iterator.next()); + + context.close(); + } + + @Test + public void propertyResolvingToExpressionTestBean() { + ConfigurableApplicationContext context = new AnnotationConfigApplicationContext( + Config.class, PropertyResolvingToExpressionTestBean.class); + + RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class); + assertEquals("one container should have been registered", 1, factory.getListenerContainers().size()); + RabbitListenerEndpoint endpoint = factory.getListenerContainers().get(0).getEndpoint(); + final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator(); + assertEquals("testQueue", iterator.next()); + assertEquals("secondQueue", iterator.next()); + + context.close(); + } + + @Test + public void invalidValueInAnnotationTestBean() { + try { + new AnnotationConfigApplicationContext(Config.class, InvalidValueInAnnotationTestBean.class); + } + catch (BeanCreationException e) { + assertThat(e.getCause(), Matchers.instanceOf(IllegalArgumentException.class)); + assertThat(e.getMessage(), Matchers.allOf( + Matchers.containsString("@RabbitListener can't resolve"), + Matchers.containsString("as either a String or a Queue") + )); + } + } @Component static class SimpleMessageListenerTestBean { @@ -93,6 +202,15 @@ public class RabbitListenerAnnotationBeanPostProcessorTests { } + @Component + static class SimpleMessageListenerWithMixedAnnotationsTestBean { + + @RabbitListener(queues = {"testQueue", "#{mySecondQueue}"}) + public void handleIt(String body) { + } + + } + @Component static class MetaAnnotationTestBean { @@ -109,8 +227,48 @@ public class RabbitListenerAnnotationBeanPostProcessorTests { static @interface FooListener { } + @Component + static class MultipleQueueNamesTestBean { + + @RabbitListener(queues = {"metaTestQueue", "#{@myTestQueue.name}"}) + public void handleIt(String body) { + } + } + + @Component + static class MultipleQueuesTestBean { + + @RabbitListener(queues = {"#{@myTestQueue}", "#{@mySecondQueue}"}) + public void handleIt(String body) { + } + } + + @Component + static class MixedQueuesAndQueueNamesTestBean { + + @RabbitListener(queues = {"metaTestQueue", "#{@myTestQueue}", "#{@mySecondQueue.name}"}) + public void handleIt(String body) { + } + } + + @Component + static class PropertyResolvingToExpressionTestBean { + + @RabbitListener(queues = {"${myQueueExpression}", "#{@mySecondQueue}"}) + public void handleIt(String body) { + } + } + + @Component + static class InvalidValueInAnnotationTestBean { + + @RabbitListener(queues = "#{@testFactory}") + public void handleIt(String body) { + } + } @Configuration + @PropertySource("classpath:/org/springframework/amqp/rabbit/annotation/queue-annotation.properties") static class Config { @Bean @@ -130,6 +288,21 @@ public class RabbitListenerAnnotationBeanPostProcessorTests { public RabbitListenerContainerTestFactory testFactory() { return new RabbitListenerContainerTestFactory(); } + + @Bean + public static PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() { + return new PropertySourcesPlaceholderConfigurer(); + } + + @Bean + public Queue myTestQueue() { + return new Queue("testQueue"); + } + + @Bean + public Queue mySecondQueue() { + return new Queue("secondQueue"); + } } } diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/annotation/queue-annotation.properties b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/annotation/queue-annotation.properties new file mode 100644 index 00000000..9d0277d1 --- /dev/null +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/annotation/queue-annotation.properties @@ -0,0 +1 @@ +myQueueExpression=#{@myTestQueue}