GH-2452: Redeclare manual entities automatically (#2453)

Fixes https://github.com/spring-projects/spring-amqp/issues/2452

The queue might be declared manually by an `AmqpAdmin`
and its name simply can be used in the listener container.
When we lose a connection, we would like to have just deleted
manual anonymous queue to be redeclared.

* Introduce `AmqpAdmin.getManualDeclarables()`
* Check for queue name presence in the `manualDeclarables` as well
from a `AbstractMessageListenerContainer.attemptDeclarations()`
* Modify `DirectMessageListenerContainerIntegrationTests.testRecoverDeletedQueueGuts()`
to deal with manual declaration and auto-recovery from the container

**Cherry-pick to `2.4.x`**
This commit is contained in:
Artem Bilan
2023-05-01 13:00:51 -04:00
committed by GitHub
parent 5ccf51f4b3
commit ee681bd015
4 changed files with 47 additions and 19 deletions

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2023 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.
@@ -16,6 +16,8 @@
package org.springframework.amqp.core;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import org.springframework.lang.Nullable;
@@ -27,6 +29,7 @@ import org.springframework.lang.Nullable;
* @author Mark Pollack
* @author Dave Syer
* @author Gary Russell
* @author Artem Bilan
*/
public interface AmqpAdmin {
@@ -126,6 +129,15 @@ public interface AmqpAdmin {
@Nullable
QueueInformation getQueueInfo(String queueName);
/**
* Return the manually declared AMQP objects.
* @return the manually declared AMQP objects.
* @since 2.4.13
*/
default Map<String, Declarable> getManualDeclarables() {
return Collections.emptyMap();
}
/**
* Initialize the admin.
* @since 2.1

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -729,6 +729,16 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
this.manualDeclarables.clear();
}
/**
* Return the manually declared AMQP objects.
* @return the manually declared AMQP objects.
* @since 2.4.13
*/
@Override
public Map<String, Declarable> getManualDeclarables() {
return Collections.unmodifiableMap(this.manualDeclarables);
}
private void processDeclarables(Collection<Exchange> contextExchanges, Collection<Queue> contextQueues,
Collection<Binding> contextBindings) {

View File

@@ -618,6 +618,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
return this.beanName;
}
@Nullable
protected final ApplicationContext getApplicationContext() {
return this.applicationContext;
}
@@ -1975,14 +1976,20 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
}
private void attemptDeclarations(AmqpAdmin admin) {
ApplicationContext context = this.getApplicationContext();
ApplicationContext context = getApplicationContext();
if (context != null) {
Set<String> queueNames = getQueueNamesAsSet();
Collection<Queue> queueBeans = new LinkedHashSet<>(
Collection<Queue> queues = new LinkedHashSet<>(
context.getBeansOfType(Queue.class, false, false).values());
Map<String, Declarables> declarables = context.getBeansOfType(Declarables.class, false, false);
declarables.values().forEach(dec -> queueBeans.addAll(dec.getDeclarablesByType(Queue.class)));
for (Queue queue : queueBeans) {
declarables.values().forEach(dec -> queues.addAll(dec.getDeclarablesByType(Queue.class)));
admin.getManualDeclarables()
.values()
.stream()
.filter(Queue.class::isInstance)
.map(Queue.class::cast)
.forEach(queues::add);
for (Queue queue : queues) {
if (isMismatchedQueuesFatal() || (queueNames.contains(queue.getName()) &&
admin.getQueueProperties(queue.getName()) == null)) {
if (logger.isDebugEnabled()) {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -539,16 +539,20 @@ public class DirectMessageListenerContainerIntegrationTests {
private void testRecoverDeletedQueueGuts(boolean autoDeclare, BrokerRunningSupport brokerRunning) throws Exception {
CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf);
GenericApplicationContext context = new GenericApplicationContext();
RabbitAdmin rabbitAdmin = new RabbitAdmin(cf);
if (autoDeclare) {
GenericApplicationContext context = new GenericApplicationContext();
context.getBeanFactory().registerSingleton("foo", new Queue(Q1));
RabbitAdmin rabbitAdmin = new RabbitAdmin(cf);
rabbitAdmin.setApplicationContext(context);
context.getBeanFactory().registerSingleton("admin", rabbitAdmin);
context.refresh();
container.setApplicationContext(context);
}
container.setAutoDeclare(autoDeclare);
else {
rabbitAdmin.setRedeclareManualDeclarations(true);
rabbitAdmin.declareQueue(new Queue(Q1));
}
rabbitAdmin.setApplicationContext(context);
context.refresh();
container.setApplicationContext(context);
container.setAmqpAdmin(rabbitAdmin);
container.setQueueNames(Q1, Q2);
container.setConsumersPerQueue(2);
container.setConsumersPerQueue(2);
@@ -565,11 +569,6 @@ public class DirectMessageListenerContainerIntegrationTests {
assertThat(consumersOnQueue(Q2, 2)).isTrue();
assertThat(activeConsumerCount(container, 2)).isTrue();
assertThat(restartConsumerCount(container, 2)).isTrue();
RabbitAdmin rabbitAdmin = new RabbitAdmin(cf);
if (!autoDeclare) {
Thread.sleep(2000);
rabbitAdmin.declareQueue(new Queue(Q1));
}
assertThat(consumersOnQueue(Q1, 2)).isTrue();
assertThat(consumersOnQueue(Q2, 2)).isTrue();
assertThat(activeConsumerCount(container, 4)).isTrue();