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

@@ -122,7 +122,6 @@ in the format of `spring.cloud.stream.rabbit.default.<property>=<value>`.
 
 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 <<spring-cloud-stream-rabbit-frame-max-headroom, frameMaxHeadroom property>>.
 +
 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-<partition>` 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.
 +

pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-build</artifactId>
-		<version>2.2.0.RELEASE</version>
+		<version>2.2.1.RELEASE</version>
 		<relativePath />
 	</parent>
 	<properties>

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;

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<DeclarableCustomizer> customizers;
+
 	public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory) {
+		this(connectionFactory, Collections.emptyList());
+	}
+
+	public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory,
+			List<DeclarableCustomizer> 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<String, Object> args,
-			RabbitCommonProperties properties, boolean isDlq) {
+	private void additionalArgs(Map<String, Object> 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);
 		}

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<AbstractMessageListenerContainer> listenerContainerCustomizer,
 			@Nullable MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer,
 			@Nullable ProducerMessageHandlerCustomizer<AmqpOutboundEndpoint> producerMessageHandlerCustomizer,
-			@Nullable ConsumerEndpointCustomizer<AmqpInboundChannelAdapter> consumerCustomizer) {
+			@Nullable ConsumerEndpointCustomizer<AmqpInboundChannelAdapter> consumerCustomizer,
+			List<DeclarableCustomizer> 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<DeclarableCustomizer> customizers) {
+		return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory, customizers);
 	}
 
 	@Bean

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<BindingInfo> 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 {