From 12018d51ff10d57880e07d682bb567b56ceb97cf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 27 Jul 2023 11:41:57 -0400 Subject: [PATCH] GH-2498: Fix Manual Redeclaration With Dup. Names 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). --- .../springframework/amqp/core/AmqpAdmin.java | 12 ++++ .../amqp/rabbit/core/RabbitAdmin.java | 61 ++++++++++++------- .../AbstractMessageListenerContainer.java | 3 +- .../amqp/rabbit/core/RabbitAdminTests.java | 19 +++--- 4 files changed, 65 insertions(+), 30 deletions(-) diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java index 91156cbd..76f7e8b6 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java @@ -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 getManualDeclarables() { return Collections.emptyMap(); } + /** + * Return the manually declared AMQP objects. + * @return the manually declared AMQP objects. + * @since 2.4.15 + */ + default Set getManualDeclarableSet() { + return Collections.emptySet(); + } + /** * Initialize the admin. * @since 2.1 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index ed1b3bf8..5b957a42 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -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 manualDeclarables = Collections.synchronizedMap(new LinkedHashMap<>()); + private final Set 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> 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 iterator = this.manualDeclarables.iterator(); while (iterator.hasNext()) { - Entry 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> 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 iterator = this.manualDeclarables.iterator(); while (iterator.hasNext()) { - Entry 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 getManualDeclarables() { - return Collections.unmodifiableMap(this.manualDeclarables); + Map 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 getManualDeclarableSet() { + return Collections.unmodifiableSet(this.manualDeclarables); } private void processDeclarables(Collection contextExchanges, Collection contextQueues, diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index efb9b98d..9a938ae6 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -1924,8 +1924,7 @@ public abstract class AbstractMessageListenerContainer extends ObservableListene context.getBeansOfType(Queue.class, false, false).values()); Map 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) diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java index b36cc3a0..c1c19b49 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java @@ -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