GH-1365: Recover Manual Declarations

Resolves #1365

* Verify bean definitions are not added to `manualDeclarables`.

Author: garyrussell

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

@@ -19,10 +19,14 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -63,6 +67,7 @@
 import org.springframework.util.StringUtils;
 
 import com.rabbitmq.client.AMQP.Queue.DeclareOk;
+import com.rabbitmq.client.AMQP.Queue.DeleteOk;
 import com.rabbitmq.client.AMQP.Queue.PurgeOk;
 import com.rabbitmq.client.Channel;
 
@@ -124,6 +129,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
 
 	private final ConnectionFactory connectionFactory;
 
+	private final Map<String, Declarable> manualDeclarables = Collections.synchronizedMap(new LinkedHashMap<>());
+
 	private String beanName;
 
 	private RetryTemplate retryTemplate;
@@ -142,6 +149,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat
 
 	private boolean explicitDeclarationsOnly;
 
+	private boolean redeclareManualDeclarations;
+
 	private volatile boolean running = false;
 
 	private volatile DeclarationExceptionEvent lastDeclarationExceptionEvent;
@@ -220,6 +229,9 @@ public void declareExchange(final Exchange exchange) {
 		try {
 			this.rabbitTemplate.execute(channel -> {
 				declareExchanges(channel, exchange);
+				if (this.redeclareManualDeclarations) {
+					this.manualDeclarables.put(exchange.getName(), exchange);
+				}
 				return null;
 			});
 		}
@@ -238,6 +250,7 @@ public boolean deleteExchange(final String exchangeName) {
 
 			try {
 				channel.exchangeDelete(exchangeName);
+				removeExchangeBindings(exchangeName);
 			}
 			catch (@SuppressWarnings(UNUSED) IOException e) {
 				return false;
@@ -246,6 +259,24 @@ public boolean deleteExchange(final String exchangeName) {
 		});
 	}
 
+	private void removeExchangeBindings(final String exchangeName) {
+		this.manualDeclarables.remove(exchangeName);
+		synchronized (this.manualDeclarables) {
+			Iterator<Entry<String, Declarable>> iterator = this.manualDeclarables.entrySet().iterator();
+			while (iterator.hasNext()) {
+				Entry<String, Declarable> next = iterator.next();
+				if (next.getValue() instanceof Binding) {
+					Binding binding = (Binding) next.getValue();
+					if ((!binding.isDestinationQueue() && binding.getDestination().equals(exchangeName))
+							|| binding.getExchange().equals(exchangeName)) {
+						iterator.remove();
+					}
+				}
+			}
+		}
+	}
+
+
 	// Queue operations
 
 	/**
@@ -266,7 +297,11 @@ public String declareQueue(final Queue queue) {
 		try {
 			return this.rabbitTemplate.execute(channel -> {
 				DeclareOk[] declared = declareQueues(channel, queue);
-				return declared.length > 0 ? declared[0].getQueue() : null;
+				String result = declared.length > 0 ? declared[0].getQueue() : null;
+				if (this.redeclareManualDeclarations) {
+					this.manualDeclarables.put(result, queue);
+				}
+				return result;
 			});
 		}
 		catch (AmqpException e) {
@@ -303,6 +338,7 @@ public boolean deleteQueue(final String queueName) {
 		return this.rabbitTemplate.execute(channel -> { // NOSONAR never returns null
 			try {
 				channel.queueDelete(queueName);
+				removeQueueBindings(queueName);
 			}
 			catch (@SuppressWarnings(UNUSED) IOException e) {
 				return false;
@@ -316,11 +352,28 @@ public boolean deleteQueue(final String queueName) {
 			"Delete a queue from the broker if unused and empty (when corresponding arguments are true")
 	public void deleteQueue(final String queueName, final boolean unused, final boolean empty) {
 		this.rabbitTemplate.execute(channel -> {
-			channel.queueDelete(queueName, unused, empty);
+			DeleteOk queueDelete = channel.queueDelete(queueName, unused, empty);
+			removeQueueBindings(queueName);
 			return null;
 		});
 	}
 
+	private void removeQueueBindings(final String queueName) {
+		this.manualDeclarables.remove(queueName);
+		synchronized (this.manualDeclarables) {
+			Iterator<Entry<String, Declarable>> iterator = this.manualDeclarables.entrySet().iterator();
+			while (iterator.hasNext()) {
+				Entry<String, Declarable> next = iterator.next();
+				if (next.getValue() instanceof Binding) {
+					Binding binding = (Binding) next.getValue();
+					if (binding.isDestinationQueue() && binding.getDestination().equals(queueName)) {
+						iterator.remove();
+					}
+				}
+			}
+		}
+	}
+
 	@Override
 	@ManagedOperation(description = "Purge a queue and optionally don't wait for the purge to occur")
 	public void purgeQueue(final String queueName, final boolean noWait) {
@@ -352,6 +405,9 @@ public void declareBinding(final Binding binding) {
 		try {
 			this.rabbitTemplate.execute(channel -> {
 				declareBindings(channel, binding);
+				if (this.redeclareManualDeclarations) {
+					this.manualDeclarables.put(binding.toString(), binding);
+				}
 				return null;
 			});
 		}
@@ -377,6 +433,7 @@ public void removeBinding(final Binding binding) {
 				channel.exchangeUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
 						binding.getArguments());
 			}
+			this.manualDeclarables.remove(binding.toString());
 			return null;
 		});
 	}
@@ -444,6 +501,37 @@ public void setExplicitDeclarationsOnly(boolean explicitDeclarationsOnly) {
 		this.explicitDeclarationsOnly = explicitDeclarationsOnly;
 	}
 
+	/**
+	 * Normally, when a connection is recovered, the admin only recovers auto-delete queues,
+	 * etc, that are declared as beans in the application context. When this is true, it
+	 * will also redeclare any manually declared {@link Declarable}s via admin methods.
+	 * @return true to redeclare.
+	 * @since 2.4
+	 */
+	public boolean isRedeclareManualDeclarations() {
+		return this.redeclareManualDeclarations;
+	}
+
+	/**
+	 * Normally, when a connection is recovered, the admin only recovers auto-delete
+	 * queues, etc, that are declared as beans in the application context. When this is
+	 * true, it will also redeclare any manually declared {@link Declarable}s via admin
+	 * methods. When a queue or exhange is deleted, it will not longer be recovered, nor
+	 * will any corresponding bindings.
+	 * @param redeclareManualDeclarations true to redeclare.
+	 * @since 2.4
+	 * @see #declareQueue(Queue)
+	 * @see #declareExchange(Exchange)
+	 * @see #declareBinding(Binding)
+	 * @see #deleteQueue(String)
+	 * @see #deleteExchange(String)
+	 * @see #removeBinding(Binding)
+	 * @see #resetAllManualDeclarations()
+	 */
+	public void setRedeclareManualDeclarations(boolean redeclareManualDeclarations) {
+		this.redeclareManualDeclarations = redeclareManualDeclarations;
+	}
+
 	/**
 	 * Set a retry template for auto declarations. There is a race condition with
 	 * auto-delete, exclusive queues in that the queue might still exist for a short time,
@@ -597,7 +685,7 @@ public void initialize() {
 			}
 		}
 
-		if (exchanges.size() == 0 && queues.size() == 0 && bindings.size() == 0) {
+		if (exchanges.size() == 0 && queues.size() == 0 && bindings.size() == 0 && this.manualDeclarables.size() == 0) {
 			this.logger.debug("Nothing to declare");
 			return;
 		}
@@ -607,10 +695,36 @@ public void initialize() {
 			declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));
 			return null;
 		});
+		if (this.manualDeclarables.size() > 0) {
+			synchronized (this.manualDeclarables) {
+				this.logger.debug("Redeclaring manually declared Declarables");
+				for (Declarable dec : this.manualDeclarables.values()) {
+					if (dec instanceof Queue) {
+						declareQueue((Queue) dec);
+					}
+					else if (dec instanceof Exchange) {
+						declareExchange((Exchange) dec);
+					}
+					else {
+						declareBinding((Binding) dec);
+					}
+				}
+			}
+		}
 		this.logger.debug("Declarations finished");
 
 	}
 
+	/**
+	 * Invoke this method to prevent the admin from recovering any declarations made
+	 * by calls to {@code declare*()} methods.
+	 * @since 2.4
+	 * @see #setRedeclareManualDeclarations(boolean)
+	 */
+	public void resetAllManualDeclarations() {
+		this.manualDeclarables.clear();
+	}
+
 	private void processDeclarables(Collection<Exchange> contextExchanges, Collection<Queue> contextQueues,
 			Collection<Binding> contextBindings) {
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java

@@ -387,6 +387,67 @@ public void testLeaderLocator() throws Exception {
 		cf.destroy();
 	}
 
+	@Test
+	void manualDeclarations() {
+		CachingConnectionFactory cf = new CachingConnectionFactory(
+				RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
+		RabbitAdmin admin = new RabbitAdmin(cf);
+		GenericApplicationContext applicationContext = new GenericApplicationContext();
+		admin.setApplicationContext(applicationContext);
+		admin.setRedeclareManualDeclarations(true);
+		applicationContext.registerBean("admin", RabbitAdmin.class, () -> admin);
+		applicationContext.registerBean("beanQueue", Queue.class,
+				() -> new Queue("thisOneShouldntBeInTheManualDecs", false, true, true));
+		applicationContext.registerBean("beanEx", DirectExchange.class,
+				() -> new DirectExchange("thisOneShouldntBeInTheManualDecs", false, true));
+		applicationContext.registerBean("beanBinding", Binding.class,
+				() -> new Binding("thisOneShouldntBeInTheManualDecs", DestinationType.QUEUE,
+						"thisOneShouldntBeInTheManualDecs", "test", null));
+		applicationContext.refresh();
+		Map<?, ?> declarables = TestUtils.getPropertyValue(admin, "manualDeclarables", Map.class);
+		assertThat(declarables).hasSize(0);
+		// check the auto-configured Declarables
+		RabbitTemplate template = new RabbitTemplate(cf);
+		template.convertAndSend("thisOneShouldntBeInTheManualDecs", "test", "foo");
+		Object received = template.receiveAndConvert("thisOneShouldntBeInTheManualDecs", 5000);
+		assertThat(received).isEqualTo("foo");
+		// manual declarations
+		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.deleteQueue("test2");
+		template.execute(chan -> chan.queueDelete("test1"));
+		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");
+		received = template.receiveAndConvert("test1", 5000);
+		assertThat(received).isEqualTo("foo");
+		admin.resetAllManualDeclarations();
+		assertThat(declarables).hasSize(0);
+		cf.resetConnection();
+		admin.initialize();
+		assertThat(admin.getQueueProperties("test1")).isNull();
+		admin.declareQueue(new Queue("test1", false, true, true));
+		admin.declareExchange(new DirectExchange("ex1", false, true));
+		admin.declareBinding(new Binding("test1", DestinationType.QUEUE, "ex1", "test", null));
+		admin.declareExchange(new DirectExchange("ex2", false, true));
+		admin.declareBinding(new Binding("test1", DestinationType.QUEUE, "ex2", "test", null));
+		admin.declareBinding(new Binding("ex1", DestinationType.EXCHANGE, "ex2", "ex1", null));
+		assertThat(declarables).hasSize(6);
+		admin.deleteExchange("ex2");
+		assertThat(declarables).hasSize(3);
+		admin.deleteQueue("test1");
+		assertThat(declarables).hasSize(1);
+		admin.deleteExchange("ex1");
+		assertThat(declarables).hasSize(0);
+		cf.destroy();
+	}
+
 	@Configuration
 	public static class Config {
 

src/reference/asciidoc/amqp.adoc

@@ -5371,6 +5371,20 @@ Starting with version 2.1, anonymous queues are declared with argument `Queue.X_
 This ensures that the queue is declared on the node to which the application is connected.
 You can revert to the previous behavior by calling `queue.setLeaderLocator(null)` after constructing the instance.
 
+[[declarable-recovery]]
+===== Recovering Auto-Delete Declarations
+
+Normally, the `RabbitAdmin` (s) only recover queues/exchanges/bindings that are declared as beans in the application context; if any such declarations are auto-delete, they will be removed by the broker if the connection is lost.
+When the connection is re-established, the admin will redeclare the entities.
+Normally, entities created by calling `admin.declareQueue(...)`, `admin.declareExchange(...)` and `admin.declareBinding(...)` will not be recovered.
+
+Starting with version 2.4, the admin has a new property `redeclareManualDeclarations`; when true, the admin will recover these entities in addition to the beans in the application context.
+
+Recovery of individual declarations will not be performed if `deleteQueue(...)`, `deleteExchange(...)` or `removeBinding(...)` is called.
+Associated bindings are removed from the recoverable entities when queues and exchanges are deleted.
+
+Finally, calling `resetAllManualDeclarations()` will prevent the recovery of any previously declared entities.
+
 [[broker-events]]
 ==== Broker Event Listener
 

src/reference/asciidoc/whats-new.adoc

@@ -6,7 +6,12 @@
 This section describes the changes between version 2.4 and version 2.4.
 See <<change-history>> for changes in previous versions.
 
-==== @RabbitListener Changes
+==== `@RabbitListener` Changes
 
 `MessageProperties` is now available for argument matching.
 See <<async-annotation-driven-enable-signature>> for more information.
+
+==== `RabbitAdmin` Changes
+
+A new property `recoverManualDeclarations` allows recovery of manually declared queues/exchanges/bindings.
+See <<declarable-recovery>> for more information.