If not specified, the default container factory is used, if any. - * @return the {@link org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory} + * The bean name of the + * {@link org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory} to + * use to create the message listener container responsible to serve this endpoint. + *
+ * If not specified, the default container factory is used, if any. If a SpEL + * expression is provided ({@code #{...}}), the expression can either evaluate to a + * container factory instance or a bean name. + * @return the + * {@link org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory} * bean name. */ String containerFactory() default ""; @@ -166,12 +171,14 @@ public @interface RabbitListener { String priority() default ""; /** - * Reference to a {@link org.springframework.amqp.rabbit.core.RabbitAdmin - * RabbitAdmin}. Required if the listener is using auto-delete - * queues and those queues are configured for conditional declaration. This - * is the admin that will (re)declare those queues when the container is - * (re)started. See the reference documentation for more information. - * @return the {@link org.springframework.amqp.rabbit.core.RabbitAdmin} bean name. + * Reference to a {@link org.springframework.amqp.core.AmqpAdmin AmqpAdmin}. + * Required if the listener is using auto-delete queues and those queues are + * configured for conditional declaration. This is the admin that will (re)declare + * those queues when the container is (re)started. See the reference documentation for + * more information. If a SpEL expression is provided ({@code #{...}}) the expression + * can evaluate to an {@link org.springframework.amqp.core.AmqpAdmin} instance + * or bean name. + * @return the {@link org.springframework.amqp.core.AmqpAdmin} bean name. */ String admin() default ""; @@ -246,8 +253,10 @@ public @interface RabbitListener { String autoStartup() default ""; /** - * Set the task executor bean name to use for this listener's container; overrides - * any executor set on the container factory. + * Set the task executor bean name to use for this listener's container; overrides any + * executor set on the container factory. If a SpEL expression is provided + * ({@code #{...}}), the expression can either evaluate to a executor instance or a bean + * name. * @return the executor bean name. * @since 2.2 */ @@ -266,7 +275,9 @@ public @interface RabbitListener { /** * The bean name of a * {@link org.springframework.amqp.rabbit.listener.adapter.ReplyPostProcessor} to post - * process a response before it is sent. + * process a response before it is sent. If a SpEL expression is provided + * ({@code #{...}}), the expression can either evaluate to a post processor instance + * or a bean name. * @return the bean name. * @since 2.2.5 * @see org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener#setReplyPostProcessor(org.springframework.amqp.rabbit.listener.adapter.ReplyPostProcessor) @@ -275,7 +286,9 @@ public @interface RabbitListener { /** * Override the container factory's message converter used for this listener. - * @return the message converter bean name. + * @return the message converter bean name. If a SpEL expression is provided + * ({@code #{...}}), the expression can either evaluate to a converter instance + * or a bean name. * @since 2.3 */ String messageConverter() default ""; 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 95532ad8..744db56d 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 @@ -38,6 +38,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.Base64UrlNamingStrategy; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.Binding.DestinationType; @@ -415,20 +416,7 @@ public class RabbitListenerAnnotationBeanPostProcessor endpoint.setConcurrency(resolveExpressionAsStringOrInteger(rabbitListener.concurrency(), "concurrency")); endpoint.setBeanFactory(this.beanFactory); endpoint.setReturnExceptions(resolveExpressionAsBoolean(rabbitListener.returnExceptions())); - Object errorHandler = resolveExpression(rabbitListener.errorHandler()); - if (errorHandler instanceof RabbitListenerErrorHandler) { - endpoint.setErrorHandler((RabbitListenerErrorHandler) errorHandler); - } - else if (errorHandler instanceof String) { - String errorHandlerBeanName = (String) errorHandler; - if (StringUtils.hasText(errorHandlerBeanName)) { - endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, RabbitListenerErrorHandler.class)); - } - } - else { - throw new IllegalStateException("error handler mut be a bean name or RabbitListenerErrorHandler, not a " - + errorHandler.getClass().toString()); - } + resolveErrorHandler(endpoint, rabbitListener); String group = rabbitListener.group(); if (StringUtils.hasText(group)) { Object resolvedGroup = resolveExpression(group); @@ -465,6 +453,20 @@ public class RabbitListenerAnnotationBeanPostProcessor return declarables; } + private void resolveErrorHandler(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener) { + Object errorHandler = resolveExpression(rabbitListener.errorHandler()); + if (errorHandler instanceof RabbitListenerErrorHandler) { + endpoint.setErrorHandler((RabbitListenerErrorHandler) errorHandler); + } + else { + String errorHandlerBeanName = resolveExpressionAsString(rabbitListener.errorHandler(), "errorHandler"); + if (StringUtils.hasText(errorHandlerBeanName)) { + endpoint.setErrorHandler( + this.beanFactory.getBean(errorHandlerBeanName, RabbitListenerErrorHandler.class)); + } + } + } + private void resolveAckMode(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener) { String ackModeAttr = rabbitListener.ackMode(); if (StringUtils.hasText(ackModeAttr)) { @@ -482,16 +484,22 @@ public class RabbitListenerAnnotationBeanPostProcessor } private void resolveAdmin(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object adminTarget) { - String rabbitAdmin = resolveExpressionAsString(rabbitListener.admin(), "admin"); - if (StringUtils.hasText(rabbitAdmin)) { - Assert.state(this.beanFactory != null, "BeanFactory must be set to resolve RabbitAdmin by bean name"); - try { - endpoint.setAdmin(this.beanFactory.getBean(rabbitAdmin, RabbitAdmin.class)); - } - catch (NoSuchBeanDefinitionException ex) { - throw new BeanInitializationException("Could not register rabbit listener endpoint on [" + - adminTarget + "], no " + RabbitAdmin.class.getSimpleName() + " with id '" + - rabbitAdmin + "' was found in the application context", ex); + Object resolved = resolveExpression(rabbitListener.admin()); + if (resolved instanceof AmqpAdmin) { + endpoint.setAdmin((AmqpAdmin) resolved); + } + else { + String rabbitAdmin = resolveExpressionAsString(rabbitListener.admin(), "admin"); + if (StringUtils.hasText(rabbitAdmin)) { + Assert.state(this.beanFactory != null, "BeanFactory must be set to resolve RabbitAdmin by bean name"); + try { + endpoint.setAdmin(this.beanFactory.getBean(rabbitAdmin, RabbitAdmin.class)); + } + catch (NoSuchBeanDefinitionException ex) { + throw new BeanInitializationException("Could not register rabbit listener endpoint on [" + + adminTarget + "], no " + RabbitAdmin.class.getSimpleName() + " with id '" + + rabbitAdmin + "' was found in the application context", ex); + } } } } @@ -501,6 +509,10 @@ public class RabbitListenerAnnotationBeanPostProcessor Object factoryTarget, String beanName) { RabbitListenerContainerFactory factory = null; + Object resolved = resolveExpression(rabbitListener.containerFactory()); + if (resolved instanceof RabbitListenerContainerFactory) { + return (RabbitListenerContainerFactory) resolved; + } String containerFactoryBeanName = resolveExpressionAsString(rabbitListener.containerFactory(), "containerFactory"); if (StringUtils.hasText(containerFactoryBeanName)) { @@ -520,15 +532,21 @@ public class RabbitListenerAnnotationBeanPostProcessor private void resolveExecutor(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object execTarget, String beanName) { - String execBeanName = resolveExpressionAsString(rabbitListener.executor(), "executor"); - if (StringUtils.hasText(execBeanName)) { - assertBeanFactory(); - try { - endpoint.setTaskExecutor(this.beanFactory.getBean(execBeanName, TaskExecutor.class)); - } - catch (NoSuchBeanDefinitionException ex) { - throw new BeanInitializationException( - noBeanFoundMessage(execTarget, beanName, execBeanName, TaskExecutor.class), ex); + Object resolved = resolveExpression(rabbitListener.executor()); + if (resolved instanceof TaskExecutor) { + endpoint.setTaskExecutor((TaskExecutor) resolved); + } + else { + String execBeanName = resolveExpressionAsString(rabbitListener.executor(), "executor"); + if (StringUtils.hasText(execBeanName)) { + assertBeanFactory(); + try { + endpoint.setTaskExecutor(this.beanFactory.getBean(execBeanName, TaskExecutor.class)); + } + catch (NoSuchBeanDefinitionException ex) { + throw new BeanInitializationException( + noBeanFoundMessage(execTarget, beanName, execBeanName, TaskExecutor.class), ex); + } } } } @@ -536,15 +554,21 @@ public class RabbitListenerAnnotationBeanPostProcessor private void resolvePostProcessor(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object target, String beanName) { - String ppBeanName = resolveExpressionAsString(rabbitListener.replyPostProcessor(), "replyPostProcessor"); - if (StringUtils.hasText(ppBeanName)) { - assertBeanFactory(); - try { - endpoint.setReplyPostProcessor(this.beanFactory.getBean(ppBeanName, ReplyPostProcessor.class)); - } - catch (NoSuchBeanDefinitionException ex) { - throw new BeanInitializationException( - noBeanFoundMessage(target, beanName, ppBeanName, ReplyPostProcessor.class), ex); + Object resolved = resolveExpression(rabbitListener.replyPostProcessor()); + if (resolved instanceof ReplyPostProcessor) { + endpoint.setReplyPostProcessor((ReplyPostProcessor) resolved); + } + else { + String ppBeanName = resolveExpressionAsString(rabbitListener.replyPostProcessor(), "replyPostProcessor"); + if (StringUtils.hasText(ppBeanName)) { + assertBeanFactory(); + try { + endpoint.setReplyPostProcessor(this.beanFactory.getBean(ppBeanName, ReplyPostProcessor.class)); + } + catch (NoSuchBeanDefinitionException ex) { + throw new BeanInitializationException( + noBeanFoundMessage(target, beanName, ppBeanName, ReplyPostProcessor.class), ex); + } } } } @@ -552,15 +576,21 @@ public class RabbitListenerAnnotationBeanPostProcessor private void resolveMessageConverter(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object target, String beanName) { - String mcBeanName = resolveExpressionAsString(rabbitListener.messageConverter(), "messageConverter"); - if (StringUtils.hasText(mcBeanName)) { - assertBeanFactory(); - try { - endpoint.setMessageConverter(this.beanFactory.getBean(mcBeanName, MessageConverter.class)); - } - catch (NoSuchBeanDefinitionException ex) { - throw new BeanInitializationException( - noBeanFoundMessage(target, beanName, mcBeanName, MessageConverter.class), ex); + Object resolved = resolveExpression(rabbitListener.messageConverter()); + if (resolved instanceof MessageConverter) { + endpoint.setMessageConverter((MessageConverter) resolved); + } + else { + String mcBeanName = resolveExpressionAsString(rabbitListener.messageConverter(), "messageConverter"); + if (StringUtils.hasText(mcBeanName)) { + assertBeanFactory(); + try { + endpoint.setMessageConverter(this.beanFactory.getBean(mcBeanName, MessageConverter.class)); + } + catch (NoSuchBeanDefinitionException ex) { + throw new BeanInitializationException( + noBeanFoundMessage(target, beanName, mcBeanName, MessageConverter.class), ex); + } } } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java index b822bdc6..da80c5e2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java @@ -27,9 +27,12 @@ import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.junit.RabbitAvailable; import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition; +import org.springframework.amqp.rabbit.listener.adapter.ReplyPostProcessor; +import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.amqp.support.converter.MessageConverter; @@ -37,6 +40,7 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -117,12 +121,35 @@ public class EnableRabbitReturnTypesTests { return new CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); } + @Bean + public RabbitAdmin admin(CachingConnectionFactory cf) { + return new RabbitAdmin(cf); + } + @Bean public Jackson2JsonMessageConverter converter() { return new Jackson2JsonMessageConverter(); } - @RabbitListener(queues = "EnableRabbitReturnTypesTests.1") + @Bean + public SimpleAsyncTaskExecutor exec() { + return new SimpleAsyncTaskExecutor(); + } + + @Bean + public ReplyPostProcessor rpp() { + return (in, out) -> out; + } + + @Bean + public RabbitListenerErrorHandler rleh() { + return (amqpMessage, message, exception) -> null; + } + + @RabbitListener(queues = "EnableRabbitReturnTypesTests.1", admin = "#{@admin}", + containerFactory = "#{@rabbitListenerContainerFactory}", + executor = "#{@exec}", replyPostProcessor = "#{@rpp}", messageConverter = "#{@converter}", + errorHandler = "#{@rleh}") public One listen1(String in) { if ("3".equals(in)) { return new Three();