Merge pull request #23091 from 01045972746
* pr/23091: Polish "Add support for RabbitMQ's addressShuffleMode property" Add support for RabbitMQ's addressShuffleMode property Closes gh-23091
This commit is contained in:
commit
cf9e85ca53
|
@ -105,6 +105,7 @@ public class RabbitAutoConfiguration {
|
||||||
.getObject());
|
.getObject());
|
||||||
PropertyMapper map = PropertyMapper.get();
|
PropertyMapper map = PropertyMapper.get();
|
||||||
map.from(properties::determineAddresses).to(factory::setAddresses);
|
map.from(properties::determineAddresses).to(factory::setAddresses);
|
||||||
|
map.from(properties::getAddressShuffleMode).whenNonNull().to(factory::setAddressShuffleMode);
|
||||||
map.from(properties::isPublisherReturns).to(factory::setPublisherReturns);
|
map.from(properties::isPublisherReturns).to(factory::setPublisherReturns);
|
||||||
map.from(properties::getPublisherConfirmType).whenNonNull().to(factory::setPublisherConfirmType);
|
map.from(properties::getPublisherConfirmType).whenNonNull().to(factory::setPublisherConfirmType);
|
||||||
RabbitProperties.Cache.Channel channel = properties.getCache().getChannel();
|
RabbitProperties.Cache.Channel channel = properties.getCache().getChannel();
|
||||||
|
|
|
@ -23,6 +23,7 @@ import java.util.List;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
|
|
||||||
import org.springframework.amqp.core.AcknowledgeMode;
|
import org.springframework.amqp.core.AcknowledgeMode;
|
||||||
|
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.AddressShuffleMode;
|
||||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
|
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
|
||||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
|
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
|
||||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||||
|
@ -87,6 +88,11 @@ public class RabbitProperties {
|
||||||
*/
|
*/
|
||||||
private String addresses;
|
private String addresses;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Mode used to shuffle configured addresses.
|
||||||
|
*/
|
||||||
|
private AddressShuffleMode addressShuffleMode = AddressShuffleMode.NONE;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Requested heartbeat timeout; zero for none. If a duration suffix is not specified,
|
* Requested heartbeat timeout; zero for none. If a duration suffix is not specified,
|
||||||
* seconds will be used.
|
* seconds will be used.
|
||||||
|
@ -282,6 +288,14 @@ public class RabbitProperties {
|
||||||
this.virtualHost = "".equals(virtualHost) ? "/" : virtualHost;
|
this.virtualHost = "".equals(virtualHost) ? "/" : virtualHost;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public AddressShuffleMode getAddressShuffleMode() {
|
||||||
|
return this.addressShuffleMode;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void setAddressShuffleMode(AddressShuffleMode addressShuffleMode) {
|
||||||
|
this.addressShuffleMode = addressShuffleMode;
|
||||||
|
}
|
||||||
|
|
||||||
public Duration getRequestedHeartbeat() {
|
public Duration getRequestedHeartbeat() {
|
||||||
return this.requestedHeartbeat;
|
return this.requestedHeartbeat;
|
||||||
}
|
}
|
||||||
|
|
|
@ -1538,6 +1538,10 @@
|
||||||
"name": "spring.r2dbc.pool.validation-depth",
|
"name": "spring.r2dbc.pool.validation-depth",
|
||||||
"defaultValue": "local"
|
"defaultValue": "local"
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
"name": "spring.rabbitmq.address-shuffle-mode",
|
||||||
|
"defaultValue": "none"
|
||||||
|
},
|
||||||
{
|
{
|
||||||
"name": "spring.rabbitmq.cache.connection.mode",
|
"name": "spring.rabbitmq.cache.connection.mode",
|
||||||
"defaultValue": "channel"
|
"defaultValue": "channel"
|
||||||
|
|
|
@ -42,6 +42,7 @@ import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFac
|
||||||
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
|
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
|
||||||
import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils;
|
import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils;
|
||||||
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
|
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
|
||||||
|
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.AddressShuffleMode;
|
||||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
|
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
|
||||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
|
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
|
||||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||||
|
@ -137,12 +138,15 @@ class RabbitAutoConfigurationTests {
|
||||||
void testConnectionFactoryWithOverrides() {
|
void testConnectionFactoryWithOverrides() {
|
||||||
this.contextRunner.withUserConfiguration(TestConfiguration.class)
|
this.contextRunner.withUserConfiguration(TestConfiguration.class)
|
||||||
.withPropertyValues("spring.rabbitmq.host:remote-server", "spring.rabbitmq.port:9000",
|
.withPropertyValues("spring.rabbitmq.host:remote-server", "spring.rabbitmq.port:9000",
|
||||||
"spring.rabbitmq.username:alice", "spring.rabbitmq.password:secret",
|
"spring.rabbitmq.address-shuffle-mode=random", "spring.rabbitmq.username:alice",
|
||||||
"spring.rabbitmq.virtual_host:/vhost", "spring.rabbitmq.connection-timeout:123")
|
"spring.rabbitmq.password:secret", "spring.rabbitmq.virtual_host:/vhost",
|
||||||
|
"spring.rabbitmq.connection-timeout:123")
|
||||||
.run((context) -> {
|
.run((context) -> {
|
||||||
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
|
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
|
||||||
assertThat(connectionFactory.getHost()).isEqualTo("remote-server");
|
assertThat(connectionFactory.getHost()).isEqualTo("remote-server");
|
||||||
assertThat(connectionFactory.getPort()).isEqualTo(9000);
|
assertThat(connectionFactory.getPort()).isEqualTo(9000);
|
||||||
|
assertThat(connectionFactory).hasFieldOrPropertyWithValue("addressShuffleMode",
|
||||||
|
AddressShuffleMode.RANDOM);
|
||||||
assertThat(connectionFactory.getVirtualHost()).isEqualTo("/vhost");
|
assertThat(connectionFactory.getVirtualHost()).isEqualTo("/vhost");
|
||||||
com.rabbitmq.client.ConnectionFactory rcf = connectionFactory.getRabbitConnectionFactory();
|
com.rabbitmq.client.ConnectionFactory rcf = connectionFactory.getRabbitConnectionFactory();
|
||||||
assertThat(rcf.getConnectionTimeout()).isEqualTo(123);
|
assertThat(rcf.getConnectionTimeout()).isEqualTo(123);
|
||||||
|
|
Loading…
Reference in New Issue