diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java index cbb61174..e7be75a3 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java +++ b/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 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { 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_RETURNS); NamespaceUtils.setValueIfAttributeDefined(builder, element, REQUESTED_HEARTBEAT, "requestedHeartBeat"); NamespaceUtils.setValueIfAttributeDefined(builder, element, CONNECTION_TIMEOUT); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index c78fd1c6..0e39cb99 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 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.ObjectUtils; 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 abstract class AbstractConnectionFactory implements ConnectionFactory, Di private ApplicationEventPublisher applicationEventPublisher; + private AddressResolver addressResolver; + private volatile boolean contextStopped; /** @@ -217,6 +220,16 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di 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 abstract class AbstractConnectionFactory implements ConnectionFactory, Di 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; } @@ -512,28 +526,47 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } private com.rabbitmq.client.Connection connect(String connectionName) throws IOException, TimeoutException { - com.rabbitmq.client.Connection rabbitConnection; + if (this.addressResolver != null) { + return connectResolver(connectionName); + } if (this.addresses != null) { - List
addressesToConnect = this.addresses; - if (this.shuffleAddresses && addressesToConnect.size() > 1) { - List list = new ArrayList<>(addressesToConnect); - Collections.shuffle(list); - addressesToConnect = list; - } - if (this.logger.isInfoEnabled()) { - this.logger.info("Attempting to connect to: " + addressesToConnect); - } - rabbitConnection = this.rabbitConnectionFactory.newConnection(this.executorService, addressesToConnect, - connectionName); + return connectAddresses(connectionName); } else { - if (this.logger.isInfoEnabled()) { - this.logger.info("Attempting to connect to: " + this.rabbitConnectionFactory.getHost() - + ":" + this.rabbitConnectionFactory.getPort()); - } - rabbitConnection = this.rabbitConnectionFactory.newConnection(this.executorService, connectionName); + return connectHostPort(connectionName); } - return rabbitConnection; + } + + 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 addressesToConnect = this.addresses; + if (this.shuffleAddresses && addressesToConnect.size() > 1) { + List 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() { diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd index 80d84333..da6eb6b7 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd @@ -1428,6 +1428,18 @@ ]]> +