From 8b2cc61f7e2c1fa486f80f7974a4512b12e55425 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 14 Feb 2022 12:13:40 -0500 Subject: [PATCH] GH-1422: @RabbitListener: Fix Broker-Named Queues Resolves https://github.com/spring-projects/spring-amqp/issues/1422 Broker-named queues did not work with `@RabbitListener` because the BPP only passed the name of the queue bean, not the bean itself, into the endpoint. Support bean injection; fall back to the previous behavior if a mixture of beans and names are encountered (the containers don't support both types of configuration). Add a note to the javadoc to indicate that broker-named queues are not supported via `queuesToDeclare` and `bindings` properties; such queues must be declared as discrete beans. * Docs. **cherry-pick to `2.4.x` & `2.3.x`** --- .../rabbit/annotation/RabbitListener.java | 6 +- ...itListenerAnnotationBeanPostProcessor.java | 69 +++++++++++++++---- .../EnableRabbitIntegrationTests.java | 20 +++++- ...tenerAnnotationBeanPostProcessorTests.java | 14 ++-- src/reference/asciidoc/amqp.adoc | 3 +- 5 files changed, 87 insertions(+), 25 deletions(-) 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 59a0cd99..51c43847 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -148,6 +148,8 @@ public @interface RabbitListener { * application context, the queue will be declared on the broker with default * binding (default exchange with the queue name as the routing key). * Mutually exclusive with {@link #bindings()} and {@link #queues()}. + * NOTE: Broker-named queues cannot be declared this way, they must be defined + * as beans (with an empty string for the name). * @return the queue(s) to declare. * @see org.springframework.amqp.rabbit.listener.MessageListenerContainer * @since 2.0 @@ -186,6 +188,8 @@ public @interface RabbitListener { * Array of {@link QueueBinding}s providing the listener's queue names, together * with the exchange and optional binding information. * Mutually exclusive with {@link #queues()} and {@link #queuesToDeclare()}. + * NOTE: Broker-named queues cannot be declared this way, they must be defined + * as beans (with an empty string for the name). * @return the bindings. * @see org.springframework.amqp.rabbit.listener.MessageListenerContainer * @since 1.5 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 dc6e5f4f..74983a73 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2021 the original author or authors. + * Copyright 2014-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -414,7 +414,19 @@ public class RabbitListenerAnnotationBeanPostProcessor endpoint.setBean(bean); endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory); endpoint.setId(getEndpointId(rabbitListener)); - endpoint.setQueueNames(resolveQueues(rabbitListener, declarables)); + List resolvedQueues = resolveQueues(rabbitListener, declarables); + if (!resolvedQueues.isEmpty()) { + if (resolvedQueues.get(0) instanceof String) { + endpoint.setQueueNames(resolvedQueues.stream() + .map(o -> (String) o) + .collect(Collectors.toList()).toArray(new String[0])); + } + else { + endpoint.setQueues(resolvedQueues.stream() + .map(o -> (Queue) o) + .collect(Collectors.toList()).toArray(new Queue[0])); + } + } endpoint.setConcurrency(resolveExpressionAsStringOrInteger(rabbitListener.concurrency(), "concurrency")); endpoint.setBeanFactory(this.beanFactory); endpoint.setReturnExceptions(resolveExpressionAsBoolean(rabbitListener.returnExceptions())); @@ -625,23 +637,29 @@ public class RabbitListenerAnnotationBeanPostProcessor } } - private String[] resolveQueues(RabbitListener rabbitListener, Collection declarables) { + private List resolveQueues(RabbitListener rabbitListener, Collection declarables) { String[] queues = rabbitListener.queues(); QueueBinding[] bindings = rabbitListener.bindings(); org.springframework.amqp.rabbit.annotation.Queue[] queuesToDeclare = rabbitListener.queuesToDeclare(); - List result = new ArrayList(); + List queueNames = new ArrayList(); + List queueBeans = new ArrayList(); if (queues.length > 0) { for (int i = 0; i < queues.length; i++) { - resolveAsString(resolveExpression(queues[i]), result, true, "queues"); + resolveQueues(queues[i], queueNames, queueBeans); } } + if (!queueNames.isEmpty()) { + // revert to the previous behavior of just using the name when there is mixture of String and Queue + queueBeans.forEach(qb -> queueNames.add(qb.getName())); + queueBeans.clear(); + } if (queuesToDeclare.length > 0) { if (queues.length > 0) { throw new BeanInitializationException( "@RabbitListener can have only one of 'queues', 'queuesToDeclare', or 'bindings'"); } for (int i = 0; i < queuesToDeclare.length; i++) { - result.add(declareQueue(queuesToDeclare[i], declarables)); + queueNames.add(declareQueue(queuesToDeclare[i], declarables)); } } if (bindings.length > 0) { @@ -649,26 +667,47 @@ public class RabbitListenerAnnotationBeanPostProcessor throw new BeanInitializationException( "@RabbitListener can have only one of 'queues', 'queuesToDeclare', or 'bindings'"); } - return registerBeansForDeclaration(rabbitListener, declarables); + return Arrays.stream(registerBeansForDeclaration(rabbitListener, declarables)) + .map(s -> (Object) s) + .collect(Collectors.toList()); } - return result.toArray(new String[result.size()]); + return queueNames.isEmpty() + ? queueBeans.stream() + .map(s -> (Object) s) + .collect(Collectors.toList()) + : queueNames.stream() + .map(s -> (Object) s) + .collect(Collectors.toList()); + + } + + private void resolveQueues(String queue, List result, List queueBeans) { + resolveAsStringOrQueue(resolveExpression(queue), result, queueBeans, "queues"); } @SuppressWarnings("unchecked") - private void resolveAsString(Object resolvedValue, List result, boolean canBeQueue, String what) { + private void resolveAsStringOrQueue(Object resolvedValue, List names, @Nullable List queues, + String what) { + Object resolvedValueToUse = resolvedValue; if (resolvedValue instanceof String[]) { resolvedValueToUse = Arrays.asList((String[]) resolvedValue); } - if (canBeQueue && resolvedValueToUse instanceof Queue) { - result.add(((Queue) resolvedValueToUse).getName()); + if (queues != null && resolvedValueToUse instanceof Queue) { + if (!names.isEmpty()) { + // revert to the previous behavior of just using the name when there is mixture of String and Queue + names.add(((Queue) resolvedValueToUse).getName()); + } + else { + queues.add((Queue) resolvedValueToUse); + } } else if (resolvedValueToUse instanceof String) { - result.add((String) resolvedValueToUse); + names.add((String) resolvedValueToUse); } else if (resolvedValueToUse instanceof Iterable) { for (Object object : (Iterable) resolvedValueToUse) { - resolveAsString(object, result, canBeQueue, what); + resolveAsStringOrQueue(object, names, queues, what); } } else { @@ -676,7 +715,7 @@ public class RabbitListenerAnnotationBeanPostProcessor "@RabbitListener." + what + " can't resolve '%s' as a String[] or a String " - + (canBeQueue ? "or a Queue" : ""), + + (queues != null ? "or a Queue" : ""), resolvedValue)); } } @@ -776,7 +815,7 @@ public class RabbitListenerAnnotationBeanPostProcessor final int length = binding.key().length; routingKeys = new ArrayList<>(); for (int i = 0; i < length; ++i) { - resolveAsString(resolveExpression(binding.key()[i]), routingKeys, false, "@QueueBinding.key"); + resolveAsStringOrQueue(resolveExpression(binding.key()[i]), routingKeys, null, "@QueueBinding.key"); } } final Map bindingArguments = resolveArguments(binding.arguments()); 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 277f2b5e..850cf110 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2021 the original author or authors. + * Copyright 2014-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -60,6 +60,7 @@ import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.core.MessagePropertiesBuilder; +import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint; @@ -74,6 +75,7 @@ import org.springframework.amqp.rabbit.junit.BrokerRunningSupport; import org.springframework.amqp.rabbit.junit.LogLevels; import org.springframework.amqp.rabbit.junit.RabbitAvailable; import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler; import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; @@ -1017,6 +1019,13 @@ public class EnableRabbitIntegrationTests { })).isEqualTo("foo, myProp=bar"); } + @Test + void listenerWithBrokerNamedQueue() { + AbstractMessageListenerContainer container = + (AbstractMessageListenerContainer) this.registry.getListenerContainer("brokerNamed"); + assertThat(container.getQueueNames()[0]).startsWith("amq.gen"); + } + interface TxService { @Transactional @@ -1419,6 +1428,10 @@ public class EnableRabbitIntegrationTests { return payload + ", myProp=" + props.getHeader("myProp"); } + @RabbitListener(id = "brokerNamed", queues = "#{@brokerNamed}") + void brokerNamed(String in) { + } + } public static class JsonObject { @@ -2010,6 +2023,11 @@ public class EnableRabbitIntegrationTests { }; } + @Bean + org.springframework.amqp.core.Queue brokerNamed() { + return QueueBuilder.nonDurable("").autoDelete().exclusive().build(); + } + } @RabbitListener(bindings = @QueueBinding 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 b7a06e28..e58b7cba 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -187,9 +187,9 @@ public class RabbitListenerAnnotationBeanPostProcessorTests { RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class); assertThat(factory.getListenerContainers().size()).as("one container should have been registered").isEqualTo(1); RabbitListenerEndpoint endpoint = factory.getListenerContainers().get(0).getEndpoint(); - final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator(); - assertThat(iterator.next()).isEqualTo("testQueue"); - assertThat(iterator.next()).isEqualTo("secondQueue"); + final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueues().iterator(); + assertThat(iterator.next().getName()).isEqualTo("testQueue"); + assertThat(iterator.next().getName()).isEqualTo("secondQueue"); context.close(); } @@ -218,9 +218,9 @@ public class RabbitListenerAnnotationBeanPostProcessorTests { RabbitListenerContainerTestFactory factory = context.getBean(RabbitListenerContainerTestFactory.class); assertThat(factory.getListenerContainers().size()).as("one container should have been registered").isEqualTo(1); RabbitListenerEndpoint endpoint = factory.getListenerContainers().get(0).getEndpoint(); - final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator(); - assertThat(iterator.next()).isEqualTo("testQueue"); - assertThat(iterator.next()).isEqualTo("secondQueue"); + final Iterator iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueues().iterator(); + assertThat(iterator.next().getName()).isEqualTo("testQueue"); + assertThat(iterator.next().getName()).isEqualTo("secondQueue"); context.close(); } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index f0957012..05675d1b 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -2340,7 +2340,8 @@ public class MyService { In the first example, a queue `myQueue` is declared automatically (durable) together with the exchange, if needed, and bound to the exchange with the routing key. -In the second example, an anonymous (exclusive, auto-delete) queue is declared and bound. +In the second example, an anonymous (exclusive, auto-delete) queue is declared and bound; the queue name is created by the framework using the `Base64UrlNamingStrategy`. +You cannot declare broker-named queues using this technique; they need to be declared as bean definitions; see <>. Multiple `QueueBinding` entries can be provided, letting the listener listen to multiple queues. In the third example, a queue with the name retrieved from property `my.queue` is declared, if necessary, with the default binding to the default exchange using the queue name as the routing key.