GH-58, 80: Exclusive Consumer and Lazy Queues

Resolves #58
Resolves #80

Add properties to support exclusive consumers and lazy queues.

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java

@@ -149,6 +149,16 @@ public abstract class RabbitCommonProperties {
 	 */
 	private String prefix = "";
 
+	/**
+	 * True if the queue is provisioned as a lazy queue.
+	 */
+	private boolean lazy;
+
+	/**
+	 * True if the DLQ is provisioned as a lazy queue.
+	 */
+	private boolean dlqLazy;
+
 	public String getExchangeType() {
 		return this.exchangeType;
 	}
@@ -342,4 +352,20 @@ public abstract class RabbitCommonProperties {
 		this.prefix = prefix;
 	}
 
+	public boolean isLazy() {
+		return this.lazy;
+	}
+
+	public void setLazy(boolean lazy) {
+		this.lazy = lazy;
+	}
+
+	public boolean isDlqLazy() {
+		return this.dlqLazy;
+	}
+
+	public void setDlqLazy(boolean dlqLazy) {
+		this.dlqLazy = dlqLazy;
+	}
+
 }

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java

@@ -50,6 +50,11 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 
 	private long recoveryInterval = 5000;
 
+	/**
+	 * True if the consumer is exclusive.
+	 */
+	private boolean exclusive;
+
 	public boolean isTransacted() {
 		return transacted;
 	}
@@ -159,4 +164,13 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	public void setRecoveryInterval(long recoveryInterval) {
 		this.recoveryInterval = recoveryInterval;
 	}
+
+	public boolean isExclusive() {
+		return this.exclusive;
+	}
+
+	public void setExclusive(boolean exclusive) {
+		this.exclusive = exclusive;
+	}
+
 }

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java

@@ -349,7 +349,7 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 				args.put("x-dead-letter-routing-key", dlRk);
 			}
 			additionalArgs(args, properties.getExpires(), properties.getMaxLength(), properties.getMaxLengthBytes(),
-					properties.getMaxPriority(), properties.getTtl());
+					properties.getMaxPriority(), properties.getTtl(), properties.isLazy());
 		}
 		else {
 			if (properties.getDlqDeadLetterExchange() != null) {
@@ -359,13 +359,14 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 				args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey());
 			}
 			additionalArgs(args, properties.getDlqExpires(), properties.getDlqMaxLength(),
-					properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl());
+					properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl(),
+					properties.isDlqLazy());
 		}
 		return args;
 	}
 
 	private void additionalArgs(Map<String, Object> args, Integer expires, Integer maxLength, Integer maxLengthBytes,
-								Integer maxPriority, Integer ttl) {
+								Integer maxPriority, Integer ttl, boolean lazy) {
 		if (expires != null) {
 			args.put("x-expires", expires);
 		}
@@ -381,6 +382,9 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<Exte
 		if (ttl != null) {
 			args.put("x-message-ttl", ttl);
 		}
+		if (lazy) {
+			args.put("x-queue-mode", "lazy");
+		}
 	}
 
 

spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc

@@ -148,6 +148,12 @@ dlqExpires::
   how long before an unused dead letter queue is deleted (ms)
 +
 Default: `no expiration`
+dlqLazy::
+  Declare the dead letter queue with the `x-queue-mode=lazy` argument.
+  See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues].
+  Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.
++
+Default: `false`.
 dlqMaxLength::
   maximum number of messages in the dead letter queue
 +
@@ -181,6 +187,11 @@ exchangeType::
   The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations.
 +
 Default: `topic`.
+exclusive::
+  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 will attempt to consume.
++
+Default: `false`.
 expires::
   how long before an unused queue is deleted (ms)
 +
@@ -189,6 +200,12 @@ headerPatterns::
   Patterns for headers to be mapped from inbound messages.
 +
 Default: `['*']` (all headers).
+lazy::
+  Declare the queue with the `x-queue-mode=lazy` argument.
+  See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues].
+  Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.
++
+Default: `false`.
 maxConcurrency::
   the maximum number of consumers
 +
@@ -327,6 +344,12 @@ dlqExpires::
   Only applies if `requiredGroups` are provided and then only to those groups.
 +
 Default: `no expiration`
+dlqLazy::
+  Declare the dead letter queue with the `x-queue-mode=lazy` argument.
+  See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues].
+  Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.
+  Only applies if `requiredGroups` are provided and then only to those groups.
++
 dlqMaxLength::
   maximum number of messages in the dead letter queue
   Only applies if `requiredGroups` are provided and then only to those groups.
@@ -368,6 +391,13 @@ headerPatterns::
   Patterns for headers to be mapped to outbound messages.
 +
 Default: `['*']` (all headers).
+lazy::
+  Declare the queue with the `x-queue-mode=lazy` argument.
+  See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues].
+  Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.
+  Only applies if `requiredGroups` are provided and then only to those groups.
++
+Default: `false`.
 maxLength::
   maximum number of messages in the queue
   Only applies if `requiredGroups` are provided and then only to those groups.

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -263,6 +263,7 @@ public class RabbitMessageChannelBinder
 		listenerContainer.setQueueNames(consumerDestination.getName());
 		listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
 		listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
+		listenerContainer.setExclusive(properties.getExtension().isExclusive());
 		listenerContainer.afterPropertiesSet();
 
 		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -16,11 +16,6 @@
 
 package org.springframework.cloud.stream.binder.rabbit;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -85,6 +80,11 @@ import org.springframework.retry.support.RetryTemplate;
 
 import com.rabbitmq.http.client.domain.QueueInfo;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
 /**
  * @author Mark Fisher
  * @author Gary Russell
@@ -164,6 +164,7 @@ public class RabbitBinderTests extends
 		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
 		properties.getExtension().setRequeueRejected(true);
 		properties.getExtension().setTransacted(true);
+		properties.getExtension().setExclusive(true);
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", null,
 				createBindableChannel("input", new BindingProperties()), properties);
 		Lifecycle endpoint = extractEndpoint(consumerBinding);
@@ -172,6 +173,7 @@ public class RabbitBinderTests extends
 		assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.AUTO);
 		assertThat(container.getQueueNames()[0]).startsWith(properties.getExtension().getPrefix());
 		assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(container, "exclusive", Boolean.class)).isTrue();
 		assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(1);
 		assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")).isNull();
 		assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isTrue();
@@ -291,6 +293,7 @@ public class RabbitBinderTests extends
 		extProps.setExchangeAutoDelete(true);
 		extProps.setBindingRoutingKey("foo");
 		extProps.setExpires(30_000);
+		extProps.setLazy(true);
 		extProps.setMaxLength(10_000);
 		extProps.setMaxLengthBytes(100_000);
 		extProps.setMaxPriority(10);
@@ -302,6 +305,7 @@ public class RabbitBinderTests extends
 		extProps.setDlqDeadLetterExchange("propsUser3");
 		extProps.setDlqDeadLetterRoutingKey("propsUser3");
 		extProps.setDlqExpires(60_000);
+		extProps.setDlqLazy(true);
 		extProps.setDlqMaxLength(20_000);
 		extProps.setDlqMaxLengthBytes(40_000);
 		extProps.setDlqMaxPriority(8);
@@ -353,6 +357,7 @@ public class RabbitBinderTests extends
 		assertThat(args.get("x-message-ttl")).isEqualTo(2_000);
 		assertThat(args.get("x-dead-letter-exchange")).isEqualTo("customDLX");
 		assertThat(args.get("x-dead-letter-routing-key")).isEqualTo("customDLRK");
+		assertThat(args.get("x-queue-mode")).isEqualTo("lazy");
 
 		queue = rmt.getClient().getQueue("/", "customDLQ");
 
@@ -370,6 +375,7 @@ public class RabbitBinderTests extends
 		assertThat(args.get("x-message-ttl")).isEqualTo(1_000);
 		assertThat(args.get("x-dead-letter-exchange")).isEqualTo("propsUser3");
 		assertThat(args.get("x-dead-letter-routing-key")).isEqualTo("propsUser3");
+		assertThat(args.get("x-queue-mode")).isEqualTo("lazy");
 	}
 
 	@Test