GH-1198: Support AddressResolver

Resolves #1198

**cherry-pick to 2.2.x, 2.1.x**

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java

@@ -42,6 +42,8 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
 
 	private static final String SHUFFLE_ADDRESSES = "shuffle-addresses";
 
+	private static final String ADDRESS_RESOLVER = "address-resolver";
+
 	private static final String VIRTUAL_HOST_ATTRIBUTE = "virtual-host";
 
 	private static final String USER_ATTRIBUTE = "username";
@@ -101,6 +103,7 @@ protected void doParse(Element element, ParserContext parserContext, BeanDefinit
 		NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE);
 		NamespaceUtils.setValueIfAttributeDefined(builder, element, ADDRESSES);
 		NamespaceUtils.setValueIfAttributeDefined(builder, element, SHUFFLE_ADDRESSES);
+		NamespaceUtils.setReferenceIfAttributeDefined(builder, element, ADDRESS_RESOLVER);
 		NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_CONFIRMS);
 		NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_RETURNS);
 		NamespaceUtils.setValueIfAttributeDefined(builder, element, REQUESTED_HEARTBEAT, "requestedHeartBeat");

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -51,6 +51,7 @@
 import org.springframework.util.StringUtils;
 
 import com.rabbitmq.client.Address;
+import com.rabbitmq.client.AddressResolver;
 import com.rabbitmq.client.BlockedListener;
 import com.rabbitmq.client.Recoverable;
 import com.rabbitmq.client.RecoveryListener;
@@ -122,6 +123,8 @@ public void handleRecovery(Recoverable recoverable) {
 
 	private ApplicationEventPublisher applicationEventPublisher;
 
+	private AddressResolver addressResolver;
+
 	private volatile boolean contextStopped;
 
 	/**
@@ -217,6 +220,16 @@ public void setConnectionThreadFactory(ThreadFactory threadFactory) {
 		this.rabbitConnectionFactory.setThreadFactory(threadFactory);
 	}
 
+	/**
+	 * Set an {@link AddressResolver} to use when creating connections; overrides
+	 * {@link #setAddresses(String)}, {@link #setHost(String)}, and {@link #setPort(int)}.
+	 * @param addressResolver the resolver.
+	 * @since 2.1.15
+	 */
+	public void setAddressResolver(AddressResolver addressResolver) {
+		this.addressResolver = addressResolver;
+	}
+
 	/**
 	 * @param uri the URI
 	 * @since 1.5
@@ -292,7 +305,8 @@ public void setAddresses(String addresses) {
 				return;
 			}
 		}
-		this.logger.info("setAddresses() called with an empty value, will be using the host+port properties for connections");
+		this.logger.info("setAddresses() called with an empty value, will be using the host+port "
+				+ " or addressResolver properties for connections");
 		this.addresses = null;
 	}
 
@@ -531,6 +545,50 @@ public void handleRecovery(Recoverable recoverable) {
 		}
 	}
 
+	private com.rabbitmq.client.Connection connect(String connectionName) throws IOException, TimeoutException {
+		if (this.addressResolver != null) {
+			return connectResolver(connectionName);
+		}
+		if (this.addresses != null) {
+			return connectAddresses(connectionName);
+		}
+		else {
+			return connectHostPort(connectionName);
+		}
+	}
+
+	private com.rabbitmq.client.Connection connectResolver(String connectionName) throws IOException, TimeoutException {
+		if (this.logger.isInfoEnabled()) {
+			this.logger.info("Attempting to connect with: " + this.addressResolver);
+		}
+		return this.rabbitConnectionFactory.newConnection(this.executorService, this.addressResolver,
+				connectionName);
+	}
+
+	private com.rabbitmq.client.Connection connectAddresses(String connectionName)
+			throws IOException, TimeoutException {
+
+		List<Address> addressesToConnect = this.addresses;
+		if (this.shuffleAddresses && addressesToConnect.size() > 1) {
+			List<Address> list = new ArrayList<>(addressesToConnect);
+			Collections.shuffle(list);
+			addressesToConnect = list;
+		}
+		if (this.logger.isInfoEnabled()) {
+			this.logger.info("Attempting to connect to: " + addressesToConnect);
+		}
+		return this.rabbitConnectionFactory.newConnection(this.executorService, addressesToConnect,
+				connectionName);
+	}
+
+	private com.rabbitmq.client.Connection connectHostPort(String connectionName) throws IOException, TimeoutException {
+		if (this.logger.isInfoEnabled()) {
+			this.logger.info("Attempting to connect to: " + this.rabbitConnectionFactory.getHost()
+					+ ":" + this.rabbitConnectionFactory.getPort());
+		}
+		return this.rabbitConnectionFactory.newConnection(this.executorService, connectionName);
+	}
+
 	protected final String getDefaultHostName() {
 		String temp;
 		try {

spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.1.xsd

@@ -1408,6 +1408,18 @@
 					]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="address-resolver" type="xsd:string" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+	An address resolver bean; overrides 'addresses' and 'host/port'.
+					]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.rabbitmq.client.AddressResolver" />
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="username" type="xsd:string" use="optional">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -129,4 +129,10 @@ public void testMultiHost() {
 				"rabbitConnectionFactory.threadFactory"));
 	}
 
+	@Test
+	void testResolver() {
+		CachingConnectionFactory connectionFactory = beanFactory.getBean("resolved", CachingConnectionFactory.class);
+		assertThat(TestUtils.getPropertyValue(connectionFactory, "addressResolver"))
+				.isSameAs(this.beanFactory.getBean("resolver"));
+	}
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java

@@ -89,6 +89,7 @@
 import org.springframework.test.util.ReflectionTestUtils;
 
 import com.rabbitmq.client.Address;
+import com.rabbitmq.client.AddressResolver;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ConfirmListener;
 import com.rabbitmq.client.ConnectionFactory;
@@ -1895,4 +1896,27 @@ public void testShuffle() throws IOException, TimeoutException {
 		assertThat(firstAddress).containsExactly("host1", "host2", "host3");
 	}
 
+	@Test
+	void testResolver() throws Exception {
+		com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
+		com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
+		Channel mockChannel = mock(Channel.class);
+
+		AddressResolver resolver = () -> Collections.singletonList(Address.parseAddress("foo:5672"));
+		when(mockConnectionFactory.newConnection(any(ExecutorService.class), eq(resolver), anyString()))
+				.thenReturn(mockConnection);
+		when(mockConnection.createChannel()).thenReturn(mockChannel);
+		when(mockChannel.isOpen()).thenReturn(true);
+		when(mockConnection.isOpen()).thenReturn(true);
+
+		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
+		ccf.setExecutor(mock(ExecutorService.class));
+		ccf.setAddressResolver(resolver);
+		Connection con = ccf.createConnection();
+		assertThat(con).isNotNull();
+		assertThat(TestUtils.getPropertyValue(con, "target", SimpleConnection.class).getDelegate())
+				.isEqualTo(mockConnection);
+		verify(mockConnectionFactory).newConnection(any(ExecutorService.class), eq(resolver), anyString());
+	}
+
 }

spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml

@@ -20,7 +20,15 @@
 
 	<rabbit:connection-factory id="native" connection-factory="connectionFactory" channel-cache-size="10" />
 
-	<bean id="connectionFactory" class="com.rabbitmq.client.ConnectionFactory"/>
+
+	<rabbit:connection-factory id="resolved" connection-factory="connectionFactory"
+		address-resolver="resolver"/>
+
+	<bean id="resolver" class="com.rabbitmq.client.ListAddressResolver">
+		<constructor-arg value="null"/>
+	</bean>
+
+	<bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean"/>
 
 	<rabbit:connection-factory id="withExecutor" host="foo" virtual-host="/bar"
 		connection-cache-size="10" port="6888" username="user" password="password"

src/reference/asciidoc/amqp.adoc

@@ -378,6 +378,11 @@ The following example with a custom thread factory that prefixes thread names wi
 ----
 ====
 
+===== AddressResolver
+
+Starting with version 2.1.15, you can now use an `AddressResover` to resolve the connection address(es).
+This will override any settings of the `addresses` and `host/port` properties.
+
 ===== Naming Connections
 
 Starting with version 1.7, a `ConnectionNameStrategy` is provided for the injection into the `AbstractionConnectionFactory`.