From 401cda1bf3a0d43e149b01800546fd766ac02099 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 2 Dec 2019 15:00:48 -0500 Subject: [PATCH] GH-274: singleActiveConsumer and arbitrary args Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/274 Add support for `x-single-active-consumer` as well as setting arbitrary arguments on queues/exchanges/bindings. Update build version; remove temporary overrides Resolves #276 --- docs/src/main/asciidoc/overview.adoc | 27 +++++++++++- pom.xml | 2 +- .../properties/RabbitCommonProperties.java | 26 +++++++++++ .../RabbitExchangeQueueProvisioner.java | 37 +++++++++++++--- ...bbitMessageChannelBinderConfiguration.java | 11 +++-- .../integration/RabbitBinderModuleTests.java | 43 +++++++++++++++++-- 6 files changed, 131 insertions(+), 15 deletions(-) diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 32e3ce452..de862eab3 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -122,7 +122,6 @@ in the format of `spring.cloud.stream.rabbit.default.=`. Also, keep in mind that binding specific property will override its equivalent in the default. - acknowledgeMode:: The acknowledge mode. + @@ -248,6 +247,10 @@ dlqQuorum.initialQuorumSize:: When `quorum.enabled=true`, set the initial quorum size. + Default: none - broker default will apply. +dlqSingleActiveConsumer:: +Set to true to set the `x-single-active-consumer` queue property to true. ++ +Default: `false` dlqTtl:: Default time to live to apply to the dead letter queue when declared (in milliseconds). + @@ -274,6 +277,7 @@ Whether to create an exclusive consumer. Concurrency should be 1 when this is `true`. Often used when strict ordering is required but enabling a hot standby instance to take over after a failure. See `recoveryInterval`, which controls how often a standby instance attempts to consume. +Consider using `singleActiveConsumer` instead when using RabbitMQ 3.8 or later. + Default: `false`. expires:: @@ -386,6 +390,10 @@ If set to `true`, the binder republishs failed messages to the DLQ with addition Also see the <>. + Default: false +singleActiveConsumer:: +Set to true to set the `x-single-active-consumer` queue property to true. ++ +Default: `false` transacted:: Whether to use transacted channels. + @@ -406,6 +414,13 @@ To set listener container properties that are not exposed as binder or binding p The binder and binding properties will be set and then the customizer will be called. The customizer (`configure()` method) is provided with the queue name as well as the consumer group as arguments. +=== Advanced Queue/Exchange/Binding Configuration + +From time to time, the RabbitMQ team add new features that are enabled by setting some argument when declaring, for example, a queue. +Generally, such features are enabled in the binder by adding appropriate properties, but this may not be immediately available in a current version. +Starting with version 3.0.1, you can now add `DeclarableCustomizer` bean(s) to the application context to modify a `Declarable` (`Queue`, `Exchange` or `Binding`) just before the declaration is performed. +This allows you to add arguments that are not currently directly supported by the binder. + [[rabbit-receiving-batch]] === Receiving Batched Messages @@ -581,6 +596,11 @@ When `quorum.enabled=true`, set the initial quorum size. Applies only when `requiredGroups` are provided and then only to those groups. + Default: none - broker default will apply. +dlqSingleActiveConsumer:: +Set to true to set the `x-single-active-consumer` queue property to true. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: `false` dlqTtl:: Default time (in milliseconds) to live to apply to the dead letter queue when declared. Applies only when `requiredGroups` are provided and then only to those groups. @@ -666,6 +686,11 @@ A SpEL expression to determine the routing key to use when publishing messages. For a fixed routing key, use a literal expression, such as `routingKeyExpression='my.routingKey'` in a properties file or `routingKeyExpression: '''my.routingKey'''` in a YAML file. + Default: `destination` or `destination-` for partitioned destinations. +singleActiveConsumer:: +Set to true to set the `x-single-active-consumer` queue property to true. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: `false` transacted:: Whether to use transacted channels. + diff --git a/pom.xml b/pom.xml index e58af51df..ecfaac140 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-build - 2.2.0.RELEASE + 2.2.1.RELEASE diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java index 318f3f646..0dd3acd30 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java @@ -221,6 +221,16 @@ public abstract class RabbitCommonProperties { */ private QuorumConfig dlqQuorum = new QuorumConfig(); + /** + * When true, set the 'x-single-active-consumer' queue argument to true. + */ + private boolean singleActiveConsumer; + + /** + * When true, set the 'x-single-active-consumer' queue argument to true. + */ + private boolean dlqSingleActiveConsumer; + public String getExchangeType() { return this.exchangeType; } @@ -510,6 +520,22 @@ public abstract class RabbitCommonProperties { this.dlqQuorum = dlqQuorum; } + public boolean isSingleActiveConsumer() { + return this.singleActiveConsumer; + } + + public void setSingleActiveConsumer(boolean singleActiveConsumer) { + this.singleActiveConsumer = singleActiveConsumer; + } + + public boolean isDlqSingleActiveConsumer() { + return this.dlqSingleActiveConsumer; + } + + public void setDlqSingleActiveConsumer(boolean dlqSingleActiveConsumer) { + this.dlqSingleActiveConsumer = dlqSingleActiveConsumer; + } + public static class QuorumConfig { private boolean enabled; diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index 336c63b02..fd7d863fd 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.binder.rabbit.provisioning; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -31,6 +32,7 @@ import org.springframework.amqp.core.Base64UrlNamingStrategy; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.Binding.DestinationType; import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.DeclarableCustomizer; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Exchange; import org.springframework.amqp.core.ExchangeBuilder; @@ -89,11 +91,20 @@ public class RabbitExchangeQueueProvisioner private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext(); + private final List customizers; + public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory) { + this(connectionFactory, Collections.emptyList()); + } + + public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory, + List customizers) { + this.rabbitAdmin = new RabbitAdmin(connectionFactory); this.autoDeclareContext.refresh(); this.rabbitAdmin.setApplicationContext(this.autoDeclareContext); this.rabbitAdmin.afterPropertiesSet(); + this.customizers = customizers; } @Override @@ -450,7 +461,11 @@ public class RabbitExchangeQueueProvisioner } } - private void declareQueue(String beanName, Queue queue) { + private void declareQueue(String beanName, Queue queueArg) { + Queue queue = queueArg; + for (DeclarableCustomizer customizer : this.customizers) { + queue = (Queue) customizer.apply(queue); + } try { this.rabbitAdmin.declareQueue(queue); } @@ -509,8 +524,7 @@ public class RabbitExchangeQueueProvisioner return args; } - private void additionalArgs(Map args, - RabbitCommonProperties properties, boolean isDlq) { + private void additionalArgs(Map args, RabbitCommonProperties properties, boolean isDlq) { Integer expires = isDlq ? properties.getDlqExpires() : properties.getExpires(); Integer maxLength = isDlq ? properties.getDlqMaxLength() : properties.getMaxLength(); @@ -523,6 +537,7 @@ public class RabbitExchangeQueueProvisioner String overflow = isDlq ? properties.getDlqOverflowBehavior() : properties.getOverflowBehavior(); QuorumConfig quorum = isDlq ? properties.getDlqQuorum() : properties.getQuorum(); + boolean singleActive = isDlq ? properties.isDlqSingleActiveConsumer() : properties.isSingleActiveConsumer(); if (expires != null) { args.put("x-expires", expires); } @@ -553,6 +568,9 @@ public class RabbitExchangeQueueProvisioner args.put("x-quorum-initial-group-size", quorum.getInitialGroupSize()); } } + if (singleActive) { + args.put("x-single-active-consumer", true); + } } public static String applyPrefix(String prefix, String name) { @@ -578,7 +596,11 @@ public class RabbitExchangeQueueProvisioner } } - private void declareExchange(final String rootName, final Exchange exchange) { + private void declareExchange(final String rootName, final Exchange exchangeArg) { + Exchange exchange = exchangeArg; + for (DeclarableCustomizer customizer : this.customizers) { + exchange = (Exchange) customizer.apply(exchange); + } try { this.rabbitAdmin.declareExchange(exchange); } @@ -610,8 +632,11 @@ public class RabbitExchangeQueueProvisioner } } - private void declareBinding(String rootName, - org.springframework.amqp.core.Binding binding) { + private void declareBinding(String rootName, org.springframework.amqp.core.Binding bindingArg) { + Binding binding = bindingArg; + for (DeclarableCustomizer customizer : this.customizers) { + binding = (Binding) customizer.apply(binding); + } try { this.rabbitAdmin.declareBinding(binding); } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index 7ceb3e262..86effae06 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -16,8 +16,10 @@ package org.springframework.cloud.stream.binder.rabbit.config; +import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import org.springframework.amqp.core.DeclarableCustomizer; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -79,11 +81,12 @@ public class RabbitMessageChannelBinderConfiguration { @Nullable ListenerContainerCustomizer listenerContainerCustomizer, @Nullable MessageSourceCustomizer sourceCustomizer, @Nullable ProducerMessageHandlerCustomizer producerMessageHandlerCustomizer, - @Nullable ConsumerEndpointCustomizer consumerCustomizer) { + @Nullable ConsumerEndpointCustomizer consumerCustomizer, + List declarableCustomizers) { RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder( this.rabbitConnectionFactory, this.rabbitProperties, - provisioningProvider(), listenerContainerCustomizer, sourceCustomizer); + provisioningProvider(declarableCustomizers), listenerContainerCustomizer, sourceCustomizer); binder.setAdminAddresses( this.rabbitBinderConfigurationProperties.getAdminAddresses()); binder.setCompressingPostProcessor(gZipPostProcessor()); @@ -109,8 +112,8 @@ public class RabbitMessageChannelBinderConfiguration { } @Bean - RabbitExchangeQueueProvisioner provisioningProvider() { - return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory); + RabbitExchangeQueueProvisioner provisioningProvider(List customizers) { + return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory, customizers); } @Bean diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index 2dc11a73b..715fabbb0 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -16,16 +16,23 @@ package org.springframework.cloud.stream.binder.rabbit.integration; +import java.net.MalformedURLException; +import java.net.URISyntaxException; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.UUID; +import com.rabbitmq.http.client.Client; +import com.rabbitmq.http.client.domain.BindingInfo; +import com.rabbitmq.http.client.domain.ExchangeInfo; +import com.rabbitmq.http.client.domain.QueueInfo; import org.junit.After; import org.junit.ClassRule; import org.junit.Test; import org.mockito.Mockito; +import org.springframework.amqp.core.DeclarableCustomizer; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -74,6 +81,7 @@ import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; import static org.mockito.BDDMockito.willReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -106,15 +114,16 @@ public class RabbitBinderModuleTests { } @Test - public void testParentConnectionFactoryInheritedByDefault() { + public void testParentConnectionFactoryInheritedByDefault() throws Exception { context = new SpringApplicationBuilder(SimpleProcessor.class) .web(WebApplicationType.NONE).run("--server.port=0", - "--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo"); + "--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo", + "--spring.cloud.stream.rabbit.bindings.input.consumer.single-active-consumer=true"); BinderFactory binderFactory = context.getBean(BinderFactory.class); Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); - ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor + CachingConnectionFactory binderConnectionFactory = (CachingConnectionFactory) binderFieldAccessor .getPropertyValue("connectionFactory"); assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class); ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); @@ -149,6 +158,27 @@ public class RabbitBinderModuleTests { "connectionNameStrategy", ConnectionNameStrategy.class); assertThat(cns.obtainNewConnectionName(cf)).isEqualTo("foo#2"); new RabbitAdmin(rabbitTestSupport.getResource()).deleteExchange("checkPF"); + checkCustomizedArgs(); + binderConnectionFactory.resetConnection(); + binderConnectionFactory.createConnection(); + checkCustomizedArgs(); + } + + private void checkCustomizedArgs() throws MalformedURLException, URISyntaxException, InterruptedException { + Client client = new Client("http://guest:guest@localhost:15672/api"); + List bindings = client.getBindingsBySource("/", "input"); + int n = 0; + while (n++ < 100 && bindings == null || bindings.size() < 1) { + Thread.sleep(100); + bindings = client.getBindingsBySource("/", "input"); + } + assertThat(bindings).isNotNull(); + assertThat(bindings.get(0).getArguments()).contains(entry("added.by", "customizer")); + ExchangeInfo exchange = client.getExchange("/", "input"); + assertThat(exchange.getArguments()).contains(entry("added.by", "customizer")); + QueueInfo queue = client.getQueue("/", bindings.get(0).getDestination()); + assertThat(queue.getArguments()).contains(entry("added.by", "customizer")); + assertThat(queue.getArguments()).contains(entry("x-single-active-consumer", Boolean.TRUE)); } @Test @@ -371,6 +401,13 @@ public class RabbitBinderModuleTests { return (producer, dest, grp) -> producer.setBeanName("setByCustomizer:" + grp); } + @Bean + public DeclarableCustomizer customizer() { + return dec -> { + dec.addArgument("added.by", "customizer"); + return dec; + }; + } } public static class ConnectionFactoryConfiguration {