Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1422: @RabbitListener: Fix Broker-Named Queues

Browse Source 
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`**
This commit is contained in:
Gary Russell
2022-02-14 12:13:40 -05:00
committed by GitHub
parent 49bc52d4fd
commit 8b2cc61f7e
5 changed files with 87 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
View File

@@ -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

69
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
View File

@@ -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<Object> 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<Declarable> declarables) {
private List<Object> resolveQueues(RabbitListener rabbitListener, Collection<Declarable> declarables) {
String[] queues = rabbitListener.queues();
QueueBinding[] bindings = rabbitListener.bindings();
org.springframework.amqp.rabbit.annotation.Queue[] queuesToDeclare = rabbitListener.queuesToDeclare();
List<String> result = new ArrayList<String>();
List<String> queueNames = new ArrayList<String>();
List<Queue> queueBeans = new ArrayList<Queue>();
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<String> result, List<Queue> queueBeans) {
resolveAsStringOrQueue(resolveExpression(queue), result, queueBeans, "queues");
}
@SuppressWarnings("unchecked")
private void resolveAsString(Object resolvedValue, List<String> result, boolean canBeQueue, String what) {
private void resolveAsStringOrQueue(Object resolvedValue, List<String> names, @Nullable List<Queue> 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<Object>) 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<String, Object> bindingArguments = resolveArguments(binding.arguments());

20
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
View File

@@ -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

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessorTests.java
View File

@@ -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<String> iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator();
assertThat(iterator.next()).isEqualTo("testQueue");
assertThat(iterator.next()).isEqualTo("secondQueue");
final Iterator<Queue> 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<String> iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueueNames().iterator();
assertThat(iterator.next()).isEqualTo("testQueue");
assertThat(iterator.next()).isEqualTo("secondQueue");
final Iterator<Queue> iterator = ((AbstractRabbitListenerEndpoint) endpoint).getQueues().iterator();
assertThat(iterator.next().getName()).isEqualTo("testQueue");
assertThat(iterator.next().getName()).isEqualTo("secondQueue");
context.close();
}

3
src/reference/asciidoc/amqp.adoc
View File

@@ -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 <<containers-and-broker-named-queues>>.
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.