AMQP-834: AnonymousQueue: Master Locator

JIRA: https://jira.spring.io/browse/AMQP-834
Resolves #818

Locate anonymous (auto-delete, exclusive) queues on the local node.

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -94,6 +94,9 @@ public AnonymousQueue(org.springframework.amqp.core.NamingStrategy namingStrateg
 	 */
 	public AnonymousQueue(org.springframework.amqp.core.NamingStrategy namingStrategy, Map<String, Object> arguments) {
 		super(namingStrategy.generateName(), false, true, true, arguments);
+		if (!getArguments().containsKey(X_QUEUE_MASTER_LOCATOR)) {
+			setMasterLocator("client-local");
+		}
 	}
 
 	/**

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

@@ -16,8 +16,10 @@
 
 package org.springframework.amqp.core;
 
+import java.util.HashMap;
 import java.util.Map;
 
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -30,6 +32,12 @@
  */
 public class Queue extends AbstractDeclarable {
 
+	/**
+	 * Argument key for the master locator.
+	 * @since 2.1
+	 */
+	public static final String X_QUEUE_MASTER_LOCATOR = "x-queue-master-locator";
+
 	private final String name;
 
 	private final boolean durable;
@@ -90,7 +98,7 @@ public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete
 		this.durable = durable;
 		this.exclusive = exclusive;
 		this.autoDelete = autoDelete;
-		this.arguments = arguments;
+		this.arguments = arguments != null ? arguments : new HashMap<>();
 	}
 
 	/**
@@ -153,6 +161,20 @@ public String getActualName() {
 		return this.actualName;
 	}
 
+	/**
+	 * Set the master locator strategy argument for this queue.
+	 * @param locator the locator; null to clear the argument.
+	 * @since 2.1
+	 */
+	public final void setMasterLocator(@Nullable String locator) {
+		if (locator == null) {
+			this.arguments.remove(X_QUEUE_MASTER_LOCATOR);
+		}
+		else {
+			this.arguments.put(X_QUEUE_MASTER_LOCATOR, locator);
+		}
+	}
+
 	@Override
 	public String toString() {
 		return "Queue [name=" + this.name + ", durable=" + this.durable + ", autoDelete=" + this.autoDelete

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -82,7 +82,8 @@ public void testUnconditional() throws Exception {
 		Channel channel = mock(Channel.class);
 		when(cf.createConnection()).thenReturn(conn);
 		when(conn.createChannel(false)).thenReturn(channel);
-		when(channel.queueDeclare("foo", true, false, false, null)).thenReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
+		when(channel.queueDeclare("foo", true, false, false, new HashMap<>()))
+			.thenReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
 		final AtomicReference<ConnectionListener> listener = new AtomicReference<ConnectionListener>();
 		doAnswer(invocation -> {
 			listener.set((ConnectionListener) invocation.getArguments()[0]);
@@ -102,7 +103,7 @@ public void testUnconditional() throws Exception {
 		assertNotNull(listener.get());
 		listener.get().onCreate(conn);
 
-		verify(channel).queueDeclare("foo", true, false, false, null);
+		verify(channel).queueDeclare("foo", true, false, false, new HashMap<>());
 		verify(channel).exchangeDeclare("bar", "direct", true, false, false, new HashMap<String, Object>());
 		verify(channel).queueBind("foo", "bar", "foo", null);
 	}
@@ -163,7 +164,8 @@ public void testUnconditionalWithExplicitFactory() throws Exception {
 		Channel channel = mock(Channel.class);
 		when(cf.createConnection()).thenReturn(conn);
 		when(conn.createChannel(false)).thenReturn(channel);
-		when(channel.queueDeclare("foo", true, false, false, null)).thenReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
+		when(channel.queueDeclare("foo", true, false, false, new HashMap<>()))
+			.thenReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
 		final AtomicReference<ConnectionListener> listener = new AtomicReference<ConnectionListener>();
 		doAnswer(invocation -> {
 			listener.set(invocation.getArgument(0));
@@ -186,7 +188,7 @@ public void testUnconditionalWithExplicitFactory() throws Exception {
 		assertNotNull(listener.get());
 		listener.get().onCreate(conn);
 
-		verify(channel).queueDeclare("foo", true, false, false, null);
+		verify(channel).queueDeclare("foo", true, false, false, new HashMap<>());
 		verify(channel).exchangeDeclare("bar", "direct", true, false, false, new HashMap<String, Object>());
 		verify(channel).queueBind("foo", "bar", "foo", null);
 	}
@@ -270,7 +272,7 @@ public void testSkipBecauseShouldntDeclare() throws Exception {
 	public void testJavaConfig() throws Exception {
 		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class);
 		Config.listener1.onCreate(Config.conn1);
-		verify(Config.channel1).queueDeclare("foo", true, false, false, null);
+		verify(Config.channel1).queueDeclare("foo", true, false, false, new HashMap<>());
 		verify(Config.channel1).exchangeDeclare("bar", "direct", true, false, true, new HashMap<String, Object>());
 		verify(Config.channel1).queueBind("foo", "bar", "foo", null);
 
@@ -357,7 +359,7 @@ public ConnectionFactory cf1() throws IOException {
 			ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
 			when(connectionFactory.createConnection()).thenReturn(conn1);
 			when(conn1.createChannel(false)).thenReturn(channel1);
-			when(channel1.queueDeclare("foo", true, false, false, null))
+			when(channel1.queueDeclare("foo", true, false, false, new HashMap<>()))
 					.thenReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
 			doAnswer(invocation -> {
 				listener1 = invocation.getArgument(0);

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

@@ -89,6 +89,8 @@
 import com.rabbitmq.client.AMQP.Queue.DeclareOk;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.http.client.Client;
+import com.rabbitmq.http.client.domain.QueueInfo;
 
 /**
  * @author Mark Pollack
@@ -107,7 +109,7 @@ public class RabbitAdminTests {
 	public ExpectedException exception = ExpectedException.none();
 
 	@Rule
-	public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+	public BrokerRunning brokerIsRunning = BrokerRunning.isBrokerAndManagementRunning();
 
 	@Test
 	public void testSettingOfNullConnectionFactory() {
@@ -372,6 +374,36 @@ public void testRetry() throws Exception {
 		ctx.close();
 	}
 
+	@Test
+	public void testMasterLocator() throws Exception {
+		CachingConnectionFactory cf = new CachingConnectionFactory(brokerIsRunning.getConnectionFactory());
+		RabbitAdmin admin = new RabbitAdmin(cf);
+		AnonymousQueue queue = new AnonymousQueue();
+		admin.declareQueue(queue);
+		Client client = new Client("http://guest:guest@localhost:15672/api");
+		QueueInfo info = client.getQueue("?", queue.getName());
+		int n = 0;
+		while (n++ < 100 && info == null) {
+			Thread.sleep(100);
+			info = client.getQueue("/", queue.getName());
+		}
+		assertNotNull(info);
+		assertThat(info.getArguments().get(Queue.X_QUEUE_MASTER_LOCATOR), equalTo("client-local"));
+
+		queue = new AnonymousQueue();
+		queue.setMasterLocator(null);
+		admin.declareQueue(queue);
+		info = client.getQueue("?", queue.getName());
+		n = 0;
+		while (n++ < 100 && info == null) {
+			Thread.sleep(100);
+			info = client.getQueue("/", queue.getName());
+		}
+		assertNotNull(info);
+		assertNull(info.getArguments().get(Queue.X_QUEUE_MASTER_LOCATOR));
+		cf.destroy();
+	}
+
 	@Configuration
 	public static class Config {
 

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

@@ -26,6 +26,7 @@
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
@@ -350,7 +351,7 @@ public void testNestedTxBinding() throws Exception {
 		given(connection.isOpen()).willReturn(true);
 		given(connection.createChannel()).willReturn(channel1, channel2);
 		DeclareOk dok = new DeclareOk("foo", 0, 0);
-		willReturn(dok).given(channel1).queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
+		willReturn(dok).given(channel1).queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), anyMap());
 		CachingConnectionFactory ccf = new CachingConnectionFactory(cf);
 		ccf.setExecutor(mock(ExecutorService.class));
 		RabbitTemplate rabbitTemplate = new RabbitTemplate(ccf);
@@ -368,7 +369,7 @@ public void testNestedTxBinding() throws Exception {
 			});
 		});
 		verify(channel1).txSelect();
-		verify(channel1).queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
+		verify(channel1).queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), anyMap());
 		assertThat(((ChannelProxy) templateChannel.get()).getTargetChannel(), equalTo(channel1));
 		verify(channel1).txCommit();
 	}

src/reference/asciidoc/amqp.adoc

@@ -3612,6 +3612,8 @@ A `<rabbit:queue/>` with an empty, or missing, `name` attribute will always crea
 
 See <<anonymous-queue>> to understand why `AnonymousQueue` is preferred over broker-generated queue names, as well as
 how to control the format of the name.
+Starting with version 2.1, anonymous queues are declared with argument `x-queue-master-locator` set to `client-local` by default.
+This ensures that the queue will be declared on the node the application is connected to.
 Declarative queues must have fixed names because they might be referenced elsewhere in the context, for example, in a
 listener:
 
@@ -3724,6 +3726,10 @@ With Spring Framework 3.2 and later, this can be declared a little more succinct
 </rabbit:queue>
 ----
 
+When using Java configuration, the `x-queue-master-locator` is supported as a first class property via the `setMasterLocator()` method on the `Queue` class.
+Starting with version 2.1, anonymous queues are declared with this property set to `client-local` by default.
+This ensures that the queue will be declared on the node the application is connected to.
+
 IMPORTANT: The RabbitMQ broker will not allow declaration of a queue with mismatched arguments.
 For example, if a `queue` already exists with no `time to live` argument, and you attempt to declare it with, say, `key="x-message-ttl" value="100"`, an exception will be thrown.
 
@@ -4148,6 +4154,10 @@ The third creates names like `custom.gen-MRBv9sqISkuCiPfOYfpo4g`.
 
 Of course, you can provide your own naming strategy bean.
 
+Starting with version 2.1, anonymous queues are declared with argument `x-queue-master-locator` set to `client-local` by default.
+This ensures that the queue will be declared on the node the application is connected to.
+To revert to the previous behavior, call `queue.setMasterLocator(null)` after constructing the instance.
+
 [[broker-events]]
 ==== Broker Event Listener
 

src/reference/asciidoc/whats-new.adoc

@@ -43,6 +43,9 @@ Users are discouraged from using the old mechanism of declaring `<Collection<Que
 By default, the old mechanism is disabled.
 See <<collection-declaration>> for more information.
 
+`AnonymousQueue` s are now declared with `x-queue-master-locator` set to `client-local` by default, to ensure the queues are created on the node the application is connected to.
+See <<broker-configuration>> for more information.
+
 ===== RabbitTemplate Changes
 
 The `RabbitTemplate` now can be configured with the `noLocalReplyConsumer` option to control a `noLocal` flag for reply consumers in the `sendAndReceive()` operations.