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

GH-2498: Fix Manual Redeclaration With Dup. Names

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/2498

Manual redeclaration logic did not account for an exchange having the
same name as a queue.

**back port to 2.4.x will have conflicts and requires instanceof polishing**
(I will do it after merge).
This commit is contained in:
Gary Russell
2023-07-27 11:41:57 -04:00
committed by GitHub
parent d1dbf10358
commit 12018d51ff
4 changed files with 65 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java
View File

@@ -19,6 +19,7 @@ package org.springframework.amqp.core;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import org.springframework.lang.Nullable;
@@ -133,11 +134,22 @@ public interface AmqpAdmin {
* Return the manually declared AMQP objects.
* @return the manually declared AMQP objects.
* @since 2.4.13
* @deprecated in favor of {@link #getManualDeclarableSet()}.
*/
@Deprecated
default Map<String, Declarable> getManualDeclarables() {
return Collections.emptyMap();
}
/**
* Return the manually declared AMQP objects.
* @return the manually declared AMQP objects.
* @since 2.4.15
*/
default Set<Declarable> getManualDeclarableSet() {
return Collections.emptySet();
}
/**
* Initialize the admin.
* @since 2.1

61
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -22,12 +22,12 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -128,7 +128,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
private final ConnectionFactory connectionFactory;
private final Map<String, Declarable> manualDeclarables = Collections.synchronizedMap(new LinkedHashMap<>());
private final Set<Declarable> manualDeclarables = Collections.synchronizedSet(new LinkedHashSet<>());
private String beanName;
@@ -229,7 +229,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
this.rabbitTemplate.execute(channel -> {
declareExchanges(channel, exchange);
if (this.redeclareManualDeclarations) {
this.manualDeclarables.put(exchange.getName(), exchange);
this.manualDeclarables.add(exchange);
}
return null;
});
@@ -259,12 +259,15 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
}
private void removeExchangeBindings(final String exchangeName) {
this.manualDeclarables.remove(exchangeName);
synchronized (this.manualDeclarables) {
Iterator<Entry<String, Declarable>> iterator = this.manualDeclarables.entrySet().iterator();
this.manualDeclarables.stream()
.filter(dec -> dec instanceof Exchange ex && ex.getName().equals(exchangeName))
.collect(Collectors.toSet())
.forEach(ex -> this.manualDeclarables.remove(ex));
Iterator<Declarable> iterator = this.manualDeclarables.iterator();
while (iterator.hasNext()) {
Entry<String, Declarable> next = iterator.next();
if (next.getValue() instanceof Binding binding &&
Declarable next = iterator.next();
if (next instanceof Binding binding &&
((!binding.isDestinationQueue() && binding.getDestination().equals(exchangeName))
|| binding.getExchange().equals(exchangeName))) {
iterator.remove();
@@ -296,7 +299,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
DeclareOk[] declared = declareQueues(channel, queue);
String result = declared.length > 0 ? declared[0].getQueue() : null;
if (this.redeclareManualDeclarations) {
this.manualDeclarables.put(result, queue);
this.manualDeclarables.add(queue);
}
return result;
});
@@ -356,12 +359,15 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
}
private void removeQueueBindings(final String queueName) {
this.manualDeclarables.remove(queueName);
synchronized (this.manualDeclarables) {
Iterator<Entry<String, Declarable>> iterator = this.manualDeclarables.entrySet().iterator();
this.manualDeclarables.stream()
.filter(dec -> dec instanceof Queue queue && queue.getName().equals(queueName))
.collect(Collectors.toSet())
.forEach(q -> this.manualDeclarables.remove(q));
Iterator<Declarable> iterator = this.manualDeclarables.iterator();
while (iterator.hasNext()) {
Entry<String, Declarable> next = iterator.next();
if (next.getValue() instanceof Binding binding &&
Declarable next = iterator.next();
if (next instanceof Binding binding &&
(binding.isDestinationQueue() && binding.getDestination().equals(queueName))) {
iterator.remove();
}
@@ -401,7 +407,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
this.rabbitTemplate.execute(channel -> {
declareBindings(channel, binding);
if (this.redeclareManualDeclarations) {
this.manualDeclarables.put(binding.toString(), binding);
this.manualDeclarables.add(binding);
}
return null;
});
@@ -703,7 +709,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
if (this.manualDeclarables.size() > 0) {
synchronized (this.manualDeclarables) {
this.logger.debug("Redeclaring manually declared Declarables");
for (Declarable dec : this.manualDeclarables.values()) {
for (Declarable dec : this.manualDeclarables) {
if (dec instanceof Queue queue) {
declareQueue(queue);
}
@@ -729,14 +735,27 @@ 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
@Deprecated
public Map<String, Declarable> getManualDeclarables() {
return Collections.unmodifiableMap(this.manualDeclarables);
Map<String, Declarable> declarables = new HashMap<>();
this.manualDeclarables.forEach(declarable -> {
if (declarable instanceof Exchange exch) {
declarables.put(exch.getName(), declarable);
}
else if (declarable instanceof Queue queue) {
declarables.put(queue.getName(), declarable);
}
else if (declarable instanceof Binding) {
declarables.put(declarable.toString(), declarable);
}
});
return declarables;
}
@Override
public Set<Declarable> getManualDeclarableSet() {
return Collections.unmodifiableSet(this.manualDeclarables);
}
private void processDeclarables(Collection<Exchange> contextExchanges, Collection<Queue> contextQueues,

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -1924,8 +1924,7 @@ public abstract class AbstractMessageListenerContainer extends ObservableListene
context.getBeansOfType(Queue.class, false, false).values());
Map<String, Declarables> declarables = context.getBeansOfType(Declarables.class, false, false);
declarables.values().forEach(dec -> queues.addAll(dec.getDeclarablesByType(Queue.class)));
admin.getManualDeclarables()
.values()
admin.getManualDeclarableSet()
.stream()
.filter(Queue.class::isInstance)
.map(Queue.class::cast)

19
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java
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.
@@ -41,6 +41,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;
@@ -401,7 +402,7 @@ public class RabbitAdminTests extends NeedsManagementTests {
() -> new Binding("thisOneShouldntBeInTheManualDecs", DestinationType.QUEUE,
"thisOneShouldntBeInTheManualDecs", "test", null));
applicationContext.refresh();
Map<?, ?> declarables = TestUtils.getPropertyValue(admin, "manualDeclarables", Map.class);
Set<?> declarables = TestUtils.getPropertyValue(admin, "manualDeclarables", Set.class);
assertThat(declarables).hasSize(0);
// check the auto-configured Declarables
RabbitTemplate template = new RabbitTemplate(cf);
@@ -409,19 +410,23 @@ public class RabbitAdminTests extends NeedsManagementTests {
Object received = template.receiveAndConvert("thisOneShouldntBeInTheManualDecs", 5000);
assertThat(received).isEqualTo("foo");
// manual declarations
admin.declareExchange(new DirectExchange("test1", false, true));
admin.declareQueue(new Queue("test1", false, true, true));
admin.declareQueue(new Queue("test2", false, true, true));
admin.declareExchange(new DirectExchange("ex1", false, true));
admin.declareBinding(new Binding("test1", DestinationType.QUEUE, "ex1", "test", null));
admin.declareBinding(new Binding("test1", DestinationType.QUEUE, "test1", "test", null));
admin.deleteQueue("test2");
template.execute(chan -> chan.queueDelete("test1"));
template.execute(chan -> {
chan.queueDelete("test1");
chan.exchangeDelete("test1");
return null;
});
cf.resetConnection();
admin.initialize();
assertThat(admin.getQueueProperties("test1")).isNotNull();
assertThat(admin.getQueueProperties("test2")).isNull();
assertThat(declarables).hasSize(3);
// verify the exchange and binding were recovered too
template.convertAndSend("ex1", "test", "foo");
template.convertAndSend("test1", "test", "foo");
received = template.receiveAndConvert("test1", 5000);
assertThat(received).isEqualTo("foo");
admin.resetAllManualDeclarations();
@@ -451,7 +456,7 @@ public class RabbitAdminTests extends NeedsManagementTests {
RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
RabbitAdmin admin = new RabbitAdmin(cf);
admin.setRedeclareManualDeclarations(true);
Map<?, ?> declarables = TestUtils.getPropertyValue(admin, "manualDeclarables", Map.class);
Set<?> declarables = TestUtils.getPropertyValue(admin, "manualDeclarables", Set.class);
assertThat(declarables).hasSize(0);
RabbitTemplate template = new RabbitTemplate(cf);
// manual declarations