AMQP-439: RabbitListener.queues: add SpEL support

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

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();

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.
 	 *

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<String> 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<String> 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<String> 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<String> 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<String> 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");
+		}
 	}
 
 }

spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/annotation/queue-annotation.properties

@@ -0,0 +1 @@
+myQueueExpression=#{@myTestQueue}