From e1580d24b0f2bf57e4b8b13356a2637ca24f1c61 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 15 May 2020 11:04:27 -0400 Subject: [PATCH] GH-1198: Support AddressResolver Resolves https://github.com/spring-projects/spring-amqp/issues/1198 **cherry-pick to 2.2.x, 2.1.x** --- .../config/ConnectionFactoryParser.java | 3 + .../connection/AbstractConnectionFactory.java | 73 ++++++++++++++----- .../amqp/rabbit/config/spring-rabbit.xsd | 12 +++ .../config/ConnectionFactoryParserTests.java | 8 +- .../CachingConnectionFactoryTests.java | 24 ++++++ .../ConnectionFactoryParserTests-context.xml | 10 ++- src/reference/asciidoc/amqp.adoc | 5 ++ 7 files changed, 113 insertions(+), 22 deletions(-) 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 @@ ]]> + + + + + + + + + + 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()); + } + } diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml index bb4ed231..6e2d7939 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml @@ -20,7 +20,15 @@ - + + + + + + + +