GH-1125: Add DeclarableCustomizer

Resolves #1125

- pull up the common `arguments` field to `AbstractDeclarable`
- add `x-single-active-consumer` as a first class `QueueBuilder` argumemt

* Doc polishing

* Add back default ctor; fix default return from removeArgument().

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java

@@ -20,7 +20,10 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -32,12 +35,32 @@
  */
 public abstract class AbstractDeclarable implements Declarable {
 
-	private volatile boolean shouldDeclare = true;
+	private boolean shouldDeclare = true;
 
-	private volatile Collection<Object> declaringAdmins = new ArrayList<Object>();
+	private Collection<Object> declaringAdmins = new ArrayList<Object>();
 
 	private boolean ignoreDeclarationExceptions;
 
+	private final Map<String, Object> arguments;
+
+	public AbstractDeclarable() {
+		this(null);
+	}
+
+	/**
+	 * Construct an instance with the supplied arguments, or an empty map if null.
+	 * @param arguments the arguments.
+	 * @since 2.2.2
+	 */
+	public AbstractDeclarable(@Nullable Map<String, Object> arguments) {
+		if (arguments != null) {
+			this.arguments = new HashMap<>(arguments);
+		}
+		else {
+			this.arguments = new HashMap<String, Object>();
+		}
+	}
+
 	@Override
 	public boolean shouldDeclare() {
 		return this.shouldDeclare;
@@ -93,4 +116,18 @@ public void setAdminsThatShouldDeclare(Object... adminArgs) {
 		this.declaringAdmins = admins;
 	}
 
+	@Override
+	public synchronized void addArgument(String argName, Object argValue) {
+		this.arguments.put(argName, argValue);
+	}
+
+	@Override
+	public synchronized Object removeArgument(String name) {
+		return this.arguments.remove(name);
+	}
+
+	public Map<String, Object> getArguments() {
+		return this.arguments;
+	}
+
 }

spring-amqp/src/main/java/org/springframework/amqp/core/AbstractExchange.java

@@ -16,7 +16,6 @@
 
 package org.springframework.amqp.core;
 
-import java.util.HashMap;
 import java.util.Map;
 
 
@@ -38,9 +37,7 @@ public abstract class AbstractExchange extends AbstractDeclarable implements Exc
 
 	private final boolean autoDelete;
 
-	private final Map<String, Object> arguments;
-
-	private volatile boolean delayed;
+	private boolean delayed;
 
 	private boolean internal;
 
@@ -75,16 +72,10 @@ public AbstractExchange(String name, boolean durable, boolean autoDelete) {
 	 * @param arguments the arguments used to declare the exchange
 	 */
 	public AbstractExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
-		super();
+		super(arguments);
 		this.name = name;
 		this.durable = durable;
 		this.autoDelete = autoDelete;
-		if (arguments != null) {
-			this.arguments = arguments;
-		}
-		else {
-			this.arguments = new HashMap<String, Object>();
-		}
 	}
 
 	@Override
@@ -105,20 +96,6 @@ public boolean isAutoDelete() {
 		return this.autoDelete;
 	}
 
-	/**
-	 * Add an argument to the arguments.
-	 * @param argName the argument name.
-	 * @param argValue the argument value.
-	 */
-	protected synchronized void addArgument(String argName, Object argValue) {
-		this.arguments.put(argName, argValue);
-	}
-
-	@Override
-	public Map<String, Object> getArguments() {
-		return this.arguments;
-	}
-
 	@Override
 	public boolean isDelayed() {
 		return this.delayed;
@@ -156,7 +133,7 @@ public String toString() {
 						 ", durable=" + this.durable +
 						 ", autoDelete=" + this.autoDelete +
 						 ", internal=" + this.internal +
-						 ", arguments="	+ this.arguments + "]";
+						 ", arguments="	+ getArguments() + "]";
 	}
 
 }

spring-amqp/src/main/java/org/springframework/amqp/core/Binding.java

@@ -18,6 +18,8 @@
 
 import java.util.Map;
 
+import org.springframework.lang.Nullable;
+
 /**
  * Simple container collecting information to describe a binding. Takes String destination and exchange names as
  * arguments to facilitate wiring using code based configuration. Can be used in conjunction with {@link AmqpAdmin}, or
@@ -54,17 +56,16 @@ public enum DestinationType {
 
 	private final String routingKey;
 
-	private final Map<String, Object> arguments;
-
 	private final DestinationType destinationType;
 
 	public Binding(String destination, DestinationType destinationType, String exchange, String routingKey,
-			Map<String, Object> arguments) {
+			@Nullable Map<String, Object> arguments) {
+
+		super(arguments);
 		this.destination = destination;
 		this.destinationType = destinationType;
 		this.exchange = exchange;
 		this.routingKey = routingKey;
-		this.arguments = arguments;
 	}
 
 	public String getDestination() {
@@ -83,18 +84,14 @@ public String getRoutingKey() {
 		return this.routingKey;
 	}
 
-	public Map<String, Object> getArguments() {
-		return this.arguments;
-	}
-
 	public boolean isDestinationQueue() {
 		return DestinationType.QUEUE.equals(this.destinationType);
 	}
 
 	@Override
 	public String toString() {
 		return "Binding [destination=" + this.destination + ", exchange=" + this.exchange + ", routingKey="
-					+ this.routingKey + "]";
+					+ this.routingKey + ", arguments=" + getArguments() + "]";
 	}
 
 }

spring-amqp/src/main/java/org/springframework/amqp/core/Declarable.java

@@ -18,6 +18,8 @@
 
 import java.util.Collection;
 
+import org.springframework.lang.Nullable;
+
 /**
  * Classes implementing this interface can be auto-declared
  * with the broker during context initialization by an {@code AmqpAdmin}.
@@ -50,4 +52,24 @@ public interface Declarable {
 	 */
 	boolean isIgnoreDeclarationExceptions();
 
+	/**
+	 * Add an argument to the declarable.
+	 * @param name the argument name.
+	 * @param value the argument value.
+	 * @since 2.2.2
+	 */
+	default void addArgument(String name, Object value) {
+		// default no-op
+	}
+
+	/**
+	 * Remove an argument from the declarable.
+	 * @param name the argument name.
+	 * @return the argument value or null if not present.
+	 * @since 2.2.2
+	 */
+	default @Nullable Object removeArgument(String name) {
+		return null;
+	}
+
 }

spring-amqp/src/main/java/org/springframework/amqp/core/DeclarableCustomizer.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.core;
+
+import java.util.function.Function;
+
+/**
+ * Beans of this type are invoked by the {@link AmqpAdmin} before declaring the
+ * {@link Declarable}, allowing customization thereof.
+ *
+ * @author Gary Russell
+ * @since 2.2.2
+ *
+ */
+@FunctionalInterface
+public interface DeclarableCustomizer extends Function<Declarable, Declarable> {
+
+}

spring-amqp/src/main/java/org/springframework/amqp/core/Queue.java

@@ -46,8 +46,6 @@ public class Queue extends AbstractDeclarable implements Cloneable {
 
 	private final boolean autoDelete;
 
-	private final Map<String, Object> arguments;
-
 	private volatile String actualName;
 
 	/**
@@ -90,15 +88,17 @@ public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete
 	 * @param autoDelete true if the server should delete the queue when it is no longer in use
 	 * @param arguments the arguments used to declare the queue
 	 */
-	public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
+	public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete,
+			@Nullable Map<String, Object> arguments) {
+
+		super(arguments);
 		Assert.notNull(name, "'name' cannot be null");
 		this.name = name;
 		this.actualName = StringUtils.hasText(name) ? name
 				: (Base64UrlNamingStrategy.DEFAULT.generateName() + "_awaiting_declaration");
 		this.durable = durable;
 		this.exclusive = exclusive;
 		this.autoDelete = autoDelete;
-		this.arguments = arguments != null ? arguments : new HashMap<>();
 	}
 
 	/**
@@ -138,10 +138,6 @@ public boolean isAutoDelete() {
 		return this.autoDelete;
 	}
 
-	public java.util.Map<java.lang.String, java.lang.Object> getArguments() {
-		return this.arguments;
-	}
-
 	/**
 	 * Set the name from the DeclareOk.
 	 * @param name the name.
@@ -168,25 +164,25 @@ public String getActualName() {
 	 */
 	public final void setMasterLocator(@Nullable String locator) {
 		if (locator == null) {
-			this.arguments.remove(X_QUEUE_MASTER_LOCATOR);
+			removeArgument(X_QUEUE_MASTER_LOCATOR);
 		}
 		else {
-			this.arguments.put(X_QUEUE_MASTER_LOCATOR, locator);
+			addArgument(X_QUEUE_MASTER_LOCATOR, locator);
 		}
 	}
 
 	@Override
 	public Object clone() {  // NOSONAR - doesn't throw CloneNotSupportedException
 		Queue queue = new Queue(this.name, this.durable, this.exclusive, // NOSONAR - doesn't need to call super.clone()
-				this.autoDelete, new HashMap<>(this.arguments));
+				this.autoDelete, new HashMap<>(getArguments()));
 		queue.setActualName(this.actualName);
 		return queue;
 	}
 
 	@Override
 	public String toString() {
 		return "Queue [name=" + this.name + ", durable=" + this.durable + ", autoDelete=" + this.autoDelete
-				+ ", exclusive=" + this.exclusive + ", arguments=" + this.arguments
+				+ ", exclusive=" + this.exclusive + ", arguments=" + getArguments()
 				+ ", actualName=" + this.actualName + "]";
 	}
 

spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java

@@ -233,6 +233,15 @@ public QueueBuilder masterLocator(MasterLocator locator) {
 		return withArgument("x-queue-master-locator", locator.getValue());
 	}
 
+	/**
+	 * Set the 'x-single-active-consumer' queue argument.
+	 * @return the builder.
+	 * @since 2.2.2
+	 */
+	public QueueBuilder singleActiveConsumer() {
+		return withArgument("x-single-active-consumer", true);
+	}
+
 	/**
 	 * Set the queue argument to declare a queue of type 'quorum' instead of 'classic'.
 	 * @return the builder.