SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-23, GH-24: Support Queue Arguments

Browse Source 
Resolves #23
Resolves #24

- support queue arguments - ttl,expires, max-length, max-length-bytes, max-priority
- also on DLQ (if auto-bound)
- also allow customization of dead letter exchange/routing key
- support DLX/DLQ configuration on the DLQ as well - combined with TTL can be used to route back to primary queue

Polishing and Document DLQ/TTL Retries Technique

Doc Polishing
This commit is contained in:
Gary Russell
2017-01-16 16:42:14 -05:00
committed by Artem Bilan
parent b654afa551
commit 19f2e349d0
7 changed files with 635 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -104,6 +104,18 @@ bindQueue::
Whether to bind the queue to the destination exchange; set to `false` if you have set up your own infrastructure and have previously created/bound the queue.
+
Default: `true`.
deadLetterQueueName::
name of the DLQ
+
Default: `prefix+destination.dlq`
deadLetterExchange::
a DLX to assign to the queue; if autoBindDlq is true
+
Default: 'prefix+DLX'
deadLetterRoutingKey::
a dead letter routing key to assign to the queue; if autoBindDlq is true
+
Default: `destination`
declareExchange::
Whether to declare the exchange for the destination.
+
@@ -113,6 +125,34 @@ delayedExchange::
The `x-delayed-type` argument is set to the `exchangeType`.
+
Default: `false`.
dlqDeadLetterExchange::
if a DLQ is declared, a DLX to assign to that queue
+
Default: `none`
dlqDeadLetterRoutingKey::
if a DLQ is declared, a dead letter routing key to assign to that queue; default none
+
Default: `none`
dlqExpires::
how long before an unused dead letter queue is deleted (ms)
+
Default: `no expiration`
dlqMaxLength::
maximum number of messages in the dead letter queue
+
Default: `no limit`
dlqMaxLengthBytes::
maximum number of total bytes in the dead letter queue from all messages
+
Default: `no limit`
dlqMaxPriority::
maximum priority of messages in the dead letter queue (0-255)
+
Default: `none`
dlqTtl::
default time to live to apply to the dead letter queue when declared (ms)
+
Default: `no limit`
durableSubscription::
Whether subscription should be durable.
Only effective if `group` is also set.
@@ -127,8 +167,26 @@ exchangeType::
The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations.
+
Default: `topic`.
expires::
how long before an unused queue is deleted (ms)
+
Default: `no expiration`
maxConcurrency::
Default: `1`.
the maximum number of consumers
+
Default: `1`.
maxLength::
maximum number of messages in the queue
+
Default: `no limit`
maxLengthBytes::
maximum number of total bytes in the queue from all messages
+
Default: `no limit`
maxPriority::
maximum priority of messages in the queue (0-255)
+
Default:: `none`
prefetch::
Prefetch count.
+
@@ -163,6 +221,10 @@ transacted::
Whether to use transacted channels.
+
Default: `false`.
ttl::
default time to live to apply to the queue when declared (ms)
+
Default: `no limit`
txSize::
The number of deliveries between acks.
+
@@ -198,6 +260,21 @@ compress::
Whether data should be compressed when sent.
+
Default: `false`.
deadLetterQueueName::
name of the DLQ
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `prefix+destination.dlq`
deadLetterExchange::
a DLX to assign to the queue; if autoBindDlq is true
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: 'prefix+DLX'
deadLetterRoutingKey::
a dead letter routing key to assign to the queue; if autoBindDlq is true
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `destination`
declareExchange::
Whether to declare the exchange for the destination.
+
@@ -215,16 +292,71 @@ deliveryMode::
Delivery mode.
+
Default: `PERSISTENT`.
dlqDeadLetterExchange::
if a DLQ is declared, a DLX to assign to that queue
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `none`
dlqDeadLetterRoutingKey::
if a DLQ is declared, a dead letter routing key to assign to that queue; default none
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `none`
dlqExpires::
how long before an unused dead letter queue is deleted (ms)
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no expiration`
dlqMaxLength::
maximum number of messages in the dead letter queue
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no limit`
dlqMaxLengthBytes::
maximum number of total bytes in the dead letter queue from all messages
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no limit`
dlqMaxPriority::
maximum priority of messages in the dead letter queue (0-255)
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `none`
dlqTtl::
default time to live to apply to the dead letter queue when declared (ms)
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no limit`
exchangeRoutingKey::
The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`).
Only applies to non-partitioned destinations.
Only applies if `requiredGroups` are provided.
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `#`.
exchangeType::
The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations.
+
Default: `topic`.
expires::
how long before an unused queue is deleted (ms)
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no expiration`
maxLength::
maximum number of messages in the queue
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no limit`
maxLengthBytes::
maximum number of total bytes in the queue from all messages
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no limit`
maxPriority::
maximum priority of messages in the queue (0-255)
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default:: `none`
prefix::
A prefix to be added to the name of the `destination` exchange.
+
@@ -245,9 +377,79 @@ transacted::
Whether to use transacted channels.
+
Default: `false`.
ttl::
default time to live to apply to the queue when declared (ms)
Only applies if `requiredGroups` are provided and then only to those groups.
+
Default: `no limit`
[NOTE]
====
In the case of RabbitMQ, content type headers can be set by external applications.
Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport (including transports, such as Kafka, that do not normally support headers).
====
== Retry With the RabbitMQ Binder
=== Overview
When retry is enabled within the binder, the listener container thread is suspended for any back off periods that are configured.
This might be important when strict ordering is required with a single consumer but for other use cases it prevents other messages from being processed on that thread.
An alternative to using binder retry is to set up dead lettering with time to live on the dead-letter queue (DLQ), as well as dead-letter configuration on the DLQ itself.
See <<rabbit-binder-properties>> for more information about the properties discussed here.
Example configuration to enable this feature:
* Set `autoBindDlq` to `true` - the binder will create a DLQ; you can optionally specify a name in `deadLetterQueueName`
* Set `dlqTtl` to the back off time you want to wait between redeliveries
* Set the `dlqDeadLetterExchange` to the default exchange - expired messages from the DLQ will be routed to the original queue since the default `deadLetterRoutingKey` is the queue name (`destination.group`)
To force a message to be dead-lettered, either throw an `AmqpRejectAndDontRequeueException`, or set `requeueRejected` to `true` and throw any exception.
The loop will continue without end, which is fine for transient problems but you may want to give up after some number of attempts.
Fortunately, RabbitMQ provides the `x-death` header which allows you to determine how many cycles have occurred.
To acknowledge a message after giving up, throw an `ImmediateAcknowledgeAmqpException`.
=== Putting it All Together
[source]
---
spring.cloud.stream.bindings.input.destination=myDestination
spring.cloud.stream.bindings.input.group=consumerGroup
#disable binder retries
spring.cloud.stream.bindings.input.consumer.max-attempts=1
#dlx/dlq setup
spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true
spring.cloud.stream.rabbit.bindings.input.consumer.dlq-ttl=5000
spring.cloud.stream.rabbit.bindings.input.consumer.dlq-dead-letter-exchange=
---
This configuration creates an exchange `myDestination` with queue `myDestination.consumerGroup` bound to a topic exchange with a wildcard routing key `#`.
It creates a DLQ bound to a direct exchange `DLX` with routing key `myDestination.consumerGroup`.
When messages are rejected, they are routed to the DLQ.
After 5 seconds, the message expires and is routed to the original queue using the queue name as the routing key.
.Spring Boot application
[source, java]
----
@SpringBootApplication
@EnableBinding(Sink.class)
public class XDeathApplication {
public static void main(String[] args) {
SpringApplication.run(XDeathApplication.class, args);
}
@StreamListener(Sink.INPUT)
public void listen(String in, @Header(name = "x-death", required = false) Map<?,?> death) {
if (death != null && death.get("count").equals(3L)) {
// giving up - don't send to DLX
throw new ImmediateAcknowledgeAmqpException("Failed after 4 attempts");
}
throw new AmqpRejectAndDontRequeueException("failed");
}
}
----
Notice that the count property in the `x-death` header is a `Long`.

226
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java
View File

@@ -16,6 +16,8 @@
package org.springframework.cloud.stream.binder.rabbit;
import org.hibernate.validator.constraints.Range;
import org.springframework.amqp.core.ExchangeTypes;
/**
@@ -46,10 +48,95 @@ public abstract class RabbitCommonProperties {
private boolean bindQueue = true;
/**
* The routing key to bind (default # for non-partitioned, destination-instanceIndex for partitioned)
* routing key to bind (default # for non-partitioned, destination-instanceIndex for partitioned)
*/
private String bindingRoutingKey;
/**
* default time to live to apply to the queue when declared (ms)
*/
private Integer ttl;
/**
* how long before an unused queue is deleted (ms)
*/
private Integer expires;
/**
* maximum number of messages in the queue
*/
private Integer maxLength;
/**
* maximum number of total bytes in the queue from all messages
*/
private Integer maxLengthBytes;
/**
* maximum priority of messages in the queue (0-255)
*/
private Integer maxPriority;
/**
* name of the DLQ - default is prefix+destination.dlq
*/
private String deadLetterQueueName;
/**
* a DLX to assign to the queue; if autoBindDlq is true, defaults to 'prefix+DLX'
*/
private String deadLetterExchange;
/**
* a dead letter routing key to assign to that queue; if autoBindDlq is true, defaults to destination
*/
private String deadLetterRoutingKey;
/**
* default time to live to apply to the dead letter queue when declared (ms)
*/
private Integer dlqTtl;
/**
* how long before an unused dead letter queue is deleted (ms)
*/
private Integer dlqExpires;
/**
* maximum number of messages in the dead letter queue
*/
private Integer dlqMaxLength;
/**
* maximum number of total bytes in the dead letter queue from all messages
*/
private Integer dlqMaxLengthBytes;
/**
* maximum priority of messages in the dead letter queue (0-255)
*/
private Integer dlqMaxPriority;
/**
* if a DLQ is declared, a DLX to assign to that queue; default none
*/
private String dlqDeadLetterExchange;
/**
* if a DLQ is declared, a dead letter routing key to assign to that queue; default none
*/
private String dlqDeadLetterRoutingKey;
/**
* true to automatically bind a dead letter queue to a DLX
*/
private boolean autoBindDlq;
/**
* prefix for elements declared in RabbitMQ (exchanges, queues)
*/
private String prefix = "";
public String getExchangeType() {
return this.exchangeType;
}
@@ -90,4 +177,141 @@ public abstract class RabbitCommonProperties {
this.bindingRoutingKey = routingKey;
}
public Integer getTtl() {
return this.ttl;
}
public void setTtl(Integer ttl) {
this.ttl = ttl;
}
public Integer getExpires() {
return this.expires;
}
public void setExpires(Integer expires) {
this.expires = expires;
}
public Integer getMaxLength() {
return this.maxLength;
}
public void setMaxLength(Integer maxLength) {
this.maxLength = maxLength;
}
public Integer getMaxLengthBytes() {
return this.maxLengthBytes;
}
public void setMaxLengthBytes(Integer maxLengthBytes) {
this.maxLengthBytes = maxLengthBytes;
}
@Range(min = 0, max = 255)
public Integer getMaxPriority() {
return this.maxPriority;
}
public void setMaxPriority(Integer maxPriority) {
this.maxPriority = maxPriority;
}
public String getDeadLetterQueueName() {
return this.deadLetterQueueName;
}
public void setDeadLetterQueueName(String deadLetterQueueName) {
this.deadLetterQueueName = deadLetterQueueName;
}
public String getDeadLetterExchange() {
return this.deadLetterExchange;
}
public void setDeadLetterExchange(String deadLetterExchange) {
this.deadLetterExchange = deadLetterExchange;
}
public String getDeadLetterRoutingKey() {
return this.deadLetterRoutingKey;
}
public void setDeadLetterRoutingKey(String deadLetterRoutingKey) {
this.deadLetterRoutingKey = deadLetterRoutingKey;
}
public Integer getDlqTtl() {
return this.dlqTtl;
}
public void setDlqTtl(Integer dlqTtl) {
this.dlqTtl = dlqTtl;
}
public Integer getDlqExpires() {
return this.dlqExpires;
}
public void setDlqExpires(Integer dlqExpires) {
this.dlqExpires = dlqExpires;
}
public Integer getDlqMaxLength() {
return this.dlqMaxLength;
}
public void setDlqMaxLength(Integer dlqMaxLength) {
this.dlqMaxLength = dlqMaxLength;
}
public Integer getDlqMaxLengthBytes() {
return this.dlqMaxLengthBytes;
}
public void setDlqMaxLengthBytes(Integer dlqMaxLengthBytes) {
this.dlqMaxLengthBytes = dlqMaxLengthBytes;
}
public Integer getDlqMaxPriority() {
return this.dlqMaxPriority;
}
public void setDlqMaxPriority(Integer dlqMaxPriority) {
this.dlqMaxPriority = dlqMaxPriority;
}
public String getDlqDeadLetterExchange() {
return this.dlqDeadLetterExchange;
}
public void setDlqDeadLetterExchange(String dlqDeadLetterExchange) {
this.dlqDeadLetterExchange = dlqDeadLetterExchange;
}
public String getDlqDeadLetterRoutingKey() {
return this.dlqDeadLetterRoutingKey;
}
public void setDlqDeadLetterRoutingKey(String dlqDeadLetterRoutingKey) {
this.dlqDeadLetterRoutingKey = dlqDeadLetterRoutingKey;
}
public boolean isAutoBindDlq() {
return autoBindDlq;
}
public void setAutoBindDlq(boolean autoBindDlq) {
this.autoBindDlq = autoBindDlq;
}
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
}

20
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java
View File

@@ -27,8 +27,6 @@ import org.springframework.util.Assert;
*/
public class RabbitConsumerProperties extends RabbitCommonProperties {
private String prefix = "";
private boolean transacted;
private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
@@ -41,8 +39,6 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
private int txSize = 1;
private boolean autoBindDlq;
private boolean durableSubscription = true;
private boolean republishToDlq;
@@ -53,14 +49,6 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
private long recoveryInterval = 5000;
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
public boolean isTransacted() {
return transacted;
}
@@ -113,14 +101,6 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
this.txSize = txSize;
}
public boolean isAutoBindDlq() {
return autoBindDlq;
}
public void setAutoBindDlq(boolean autoBindDlq) {
this.autoBindDlq = autoBindDlq;
}
public boolean isDurableSubscription() {
return durableSubscription;
}

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

@@ -23,7 +23,9 @@ import java.util.Map;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.BindingBuilder.DirectExchangeRoutingKeyConfigurer;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
@@ -227,7 +229,7 @@ public class RabbitMessageChannelBinder
if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) {
RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
.retryOperations(buildRetryTemplate(properties))
.recoverer(determineRecoverer(baseQueueName, properties.getExtension().getPrefix(),
.recoverer(determineRecoverer(baseQueueName, properties.getExtension(),
properties.getExtension().isRepublishToDlq()))
.build();
listenerContainer.setAdviceChain(retryInterceptor);
@@ -274,7 +276,7 @@ public class RabbitMessageChannelBinder
Queue queue;
if (anonymous) {
queue = new Queue(queueName, false, true, true);
queue = new Queue(queueName, false, true, true, queueArgs(queueName, properties.getExtension(), false));
}
else {
if (partitioned) {
@@ -283,11 +285,11 @@ public class RabbitMessageChannelBinder
}
if (durable) {
queue = new Queue(queueName, true, false, false,
queueArgs(queueName, properties.getExtension().getPrefix(),
properties.getExtension().isAutoBindDlq()));
queueArgs(queueName, properties.getExtension(), false));
}
else {
queue = new Queue(queueName, false, false, true);
queue = new Queue(queueName, false, false, true,
queueArgs(queueName, properties.getExtension(), false));
}
}
declareQueue(queueName, queue);
@@ -296,7 +298,7 @@ public class RabbitMessageChannelBinder
}
if (durable) {
autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName,
properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
properties.getExtension());
}
return queue;
}
@@ -306,21 +308,12 @@ public class RabbitMessageChannelBinder
: groupedName(name, group);
}
private Map<String, Object> queueArgs(String queueName, String prefix, boolean bindDlq) {
Map<String, Object> args = new HashMap<>();
if (bindDlq) {
args.put("x-dead-letter-exchange", applyPrefix(prefix, "DLX"));
args.put("x-dead-letter-routing-key", queueName);
}
return args;
}
private MessageRecoverer determineRecoverer(String name, String prefix, boolean republish) {
private MessageRecoverer determineRecoverer(String name, RabbitCommonProperties properties, boolean republish) {
if (republish) {
RabbitTemplate errorTemplate = new RabbitTemplate(this.connectionFactory);
return new RepublishMessageRecoverer(errorTemplate,
deadLetterExchangeName(prefix),
applyPrefix(prefix, name));
deadLetterExchangeName(properties),
applyPrefix(properties.getPrefix(), name));
}
else {
return new RejectAndDontRequeueRecoverer();
@@ -372,10 +365,9 @@ public class RabbitMessageChannelBinder
String baseQueueName = exchangeName + "." + requiredGroupName;
if (!properties.isPartitioned()) {
Queue queue = new Queue(baseQueueName, true, false, false,
queueArgs(baseQueueName, prefix, extendedProperties.isAutoBindDlq()));
queueArgs(baseQueueName, extendedProperties, false));
declareQueue(baseQueueName, queue);
autoBindDLQ(baseQueueName, baseQueueName, extendedProperties.getPrefix(),
extendedProperties.isAutoBindDlq());
autoBindDLQ(baseQueueName, baseQueueName, extendedProperties);
if (extendedProperties.isBindQueue()) {
notPartitionedBinding(exchange, queue, extendedProperties);
}
@@ -386,11 +378,9 @@ public class RabbitMessageChannelBinder
String partitionSuffix = "-" + i;
String partitionQueueName = baseQueueName + partitionSuffix;
Queue queue = new Queue(partitionQueueName, true, false, false,
queueArgs(partitionQueueName, extendedProperties.getPrefix(),
extendedProperties.isAutoBindDlq()));
queueArgs(partitionQueueName, extendedProperties, false));
declareQueue(queue.getName(), queue);
autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, extendedProperties.getPrefix(),
extendedProperties.isAutoBindDlq());
autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, extendedProperties);
if (extendedProperties.isBindQueue()) {
partitionedBinding(destination, exchange, queue, extendedProperties, i);
}
@@ -407,6 +397,61 @@ public class RabbitMessageChannelBinder
return endpoint;
}
private Map<String, Object> queueArgs(String queueName, RabbitCommonProperties properties, boolean isDlq) {
Map<String, Object> args = new HashMap<>();
if (!isDlq) {
if (properties.isAutoBindDlq()) {
String dlx;
if (properties.getDeadLetterExchange() != null) {
dlx = properties.getDeadLetterExchange();
}
else {
dlx = applyPrefix(properties.getPrefix(), "DLX");
}
args.put("x-dead-letter-exchange", dlx);
String dlRk;
if (properties.getDeadLetterRoutingKey() != null) {
dlRk = properties.getDeadLetterRoutingKey();
}
else {
dlRk = queueName;
}
args.put("x-dead-letter-routing-key", dlRk);
}
additionalArgs(args, properties.getExpires(), properties.getMaxLength(), properties.getMaxLengthBytes(),
properties.getMaxPriority(), properties.getTtl());
}
else {
if (properties.getDlqDeadLetterExchange() != null) {
args.put("x-dead-letter-exchange", properties.getDlqDeadLetterExchange());
}
if (properties.getDlqDeadLetterRoutingKey() != null) {
args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey());
}
additionalArgs(args, properties.getDlqExpires(), properties.getDlqMaxLength(),
properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl());
}
return args;
}
private void additionalArgs(Map<String, Object> args, Integer expires, Integer maxLength, Integer maxLengthBytes,
Integer maxPriority, Integer ttl) {
if (expires != null) {
args.put("x-expires", expires);
}
if (maxLength != null) {
args.put("x-max-length", maxLength);
}
if (maxLengthBytes != null) {
args.put("x-max-length-bytes", maxLengthBytes);
}
if (maxPriority != null) {
args.put("x-max-priority", maxPriority);
}
if (ttl != null) {
args.put("x-message-ttl", ttl);
}
}
private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) {
RabbitTemplate rabbitTemplate;
@@ -436,23 +481,44 @@ public class RabbitMessageChannelBinder
* queue name because we use default exchange routing by queue name for the original message.
* @param baseQueueName The base name for the queue (including the binder prefix, if any).
* @param routingKey The routing key for the queue.
* @param autoBindDlq true if the DLQ should be bound.
* @param properties the properties.
*/
private void autoBindDLQ(final String baseQueueName, String routingKey, String prefix, boolean autoBindDlq) {
private void autoBindDLQ(final String baseQueueName, String routingKey, RabbitCommonProperties properties) {
boolean autoBindDlq = properties.isAutoBindDlq();
if (this.logger.isDebugEnabled()) {
this.logger.debug("autoBindDLQ=" + autoBindDlq
+ " for: " + baseQueueName);
}
if (autoBindDlq) {
String dlqName = constructDLQName(baseQueueName);
Queue dlq = new Queue(dlqName);
String dlqName;
if (properties.getDeadLetterQueueName() == null) {
dlqName = constructDLQName(baseQueueName);
}
else {
dlqName = properties.getDeadLetterQueueName();
}
Queue dlq = new Queue(dlqName, true, false, false, queueArgs(dlqName, properties, true));
declareQueue(dlqName, dlq);
final String dlxName = deadLetterExchangeName(prefix);
String dlxName = deadLetterExchangeName(properties);
final DirectExchange dlx = new DirectExchange(dlxName);
declareExchange(dlxName, dlx);
declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(routingKey));
// Also bind with the base queue name in case republishToDlq is used, which does not know about partitioning
declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(baseQueueName));
DirectExchangeRoutingKeyConfigurer bindingBuilder = BindingBuilder.bind(dlq).to(dlx);
Binding dlqBinding;
if (properties.getDeadLetterRoutingKey() == null) {
dlqBinding = bindingBuilder.with(routingKey);
}
else {
dlqBinding = bindingBuilder.with(properties.getDeadLetterRoutingKey());
}
declareBinding(dlqName, dlqBinding);
if (properties instanceof RabbitConsumerProperties &&
((RabbitConsumerProperties) properties).isRepublishToDlq()) {
/*
* Also bind with the base queue name when republishToDlq is used, which does not know about
* partitioning
*/
declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(baseQueueName));
}
}
}
@@ -569,8 +635,13 @@ public class RabbitMessageChannelBinder
addToAutoDeclareContext(rootName + ".binding", binding);
}
private String deadLetterExchangeName(String prefix) {
return prefix + DEAD_LETTER_EXCHANGE;
private String deadLetterExchangeName(RabbitCommonProperties properties) {
if (properties.getDeadLetterExchange() == null) {
return properties.getPrefix() + DEAD_LETTER_EXCHANGE;
}
else {
return properties.getDeadLetterExchange();
}
}
private void addToAutoDeclareContext(String name, Object bean) {

20
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java
View File

@@ -26,12 +26,8 @@ import org.springframework.amqp.core.MessageDeliveryMode;
*/
public class RabbitProducerProperties extends RabbitCommonProperties {
private String prefix = "";
private String[] requestHeaderPatterns = new String[] {"STANDARD_REQUEST_HEADERS", "*"};
private boolean autoBindDlq;
private boolean compress;
private boolean batchingEnabled;
@@ -58,14 +54,6 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
*/
private String routingKeyExpression;
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
public void setRequestHeaderPatterns(String[] requestHeaderPatterns) {
this.requestHeaderPatterns = requestHeaderPatterns;
}
@@ -74,14 +62,6 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
return requestHeaderPatterns;
}
public void setAutoBindDlq(boolean autoBindDlq) {
this.autoBindDlq = autoBindDlq;
}
public boolean isAutoBindDlq() {
return autoBindDlq;
}
public void setCompress(boolean compress) {
this.compress = compress;
}

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

@@ -74,6 +74,8 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import com.rabbitmq.http.client.domain.QueueInfo;
/**
* @author Mark Fisher
* @author Gary Russell
@@ -259,6 +261,93 @@ public class RabbitBinderTests extends
assertThat(exchange).isInstanceOf(DirectExchange.class);
}
@Test
public void testConsumerPropertiesWithUserInfrastructureCustomQueueArgs() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
RabbitConsumerProperties extProps = properties.getExtension();
extProps.setExchangeType(ExchangeTypes.DIRECT);
extProps.setBindingRoutingKey("foo");
extProps.setExpires(30_000);
extProps.setMaxLength(10_000);
extProps.setMaxLengthBytes(100_000);
extProps.setMaxPriority(10);
extProps.setTtl(2_000);
extProps.setAutoBindDlq(true);
extProps.setDeadLetterQueueName("customDLQ");
extProps.setDeadLetterExchange("customDLX");
extProps.setDeadLetterRoutingKey("customDLRK");
extProps.setDlqDeadLetterExchange("propsUser3");
extProps.setDlqDeadLetterRoutingKey("propsUser3");
extProps.setDlqExpires(60_000);
extProps.setDlqMaxLength(20_000);
extProps.setDlqMaxLengthBytes(40_000);
extProps.setDlqMaxPriority(8);
extProps.setDlqTtl(1_000);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("propsUser3", "infra",
createBindableChannel("input", new BindingProperties()), properties);
Lifecycle endpoint = extractEndpoint(consumerBinding);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertThat(container.isRunning()).isTrue();
consumerBinding.unbind();
assertThat(container.isRunning()).isFalse();
RabbitManagementTemplate rmt = new RabbitManagementTemplate();
List<org.springframework.amqp.core.Binding> bindings = rmt.getBindingsForExchange("/", "propsUser3");
int n = 0;
while (n++ < 100 && bindings == null || bindings.size() < 1) {
Thread.sleep(100);
bindings = rmt.getBindingsForExchange("/", "propsUser3");
}
assertThat(bindings.size()).isEqualTo(1);
assertThat(bindings.get(0).getExchange()).isEqualTo("propsUser3");
assertThat(bindings.get(0).getDestination()).isEqualTo("propsUser3.infra");
assertThat(bindings.get(0).getRoutingKey()).isEqualTo("foo");
Exchange exchange = rmt.getExchange("propsUser3");
n = 0;
while (n++ < 100 && exchange == null) {
Thread.sleep(100);
exchange = rmt.getExchange("propsUser3");
}
assertThat(exchange).isInstanceOf(DirectExchange.class);
// Queue queue = rmt.getQueue("propsUser3"); AMQP-698
QueueInfo queue = rmt.getClient().getQueue("/", "propsUser3.infra");
n = 0;
while (n++ < 100 && queue == null) {
Thread.sleep(100);
queue = rmt.getClient().getQueue("/", "propsUser3.infra");
}
assertThat(queue).isNotNull();
Map<String, Object> args = queue.getArguments();
assertThat(args.get("x-expires")).isEqualTo(30_000);
assertThat(args.get("x-max-length")).isEqualTo(10_000);
assertThat(args.get("x-max-length-bytes")).isEqualTo(100_000);
assertThat(args.get("x-max-priority")).isEqualTo(10);
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");
queue = rmt.getClient().getQueue("/", "customDLQ");
n = 0;
while (n++ < 100 && queue == null) {
Thread.sleep(100);
queue = rmt.getClient().getQueue("/", "customDLQ");
}
assertThat(queue).isNotNull();
args = queue.getArguments();
assertThat(args.get("x-expires")).isEqualTo(60_000);
assertThat(args.get("x-max-length")).isEqualTo(20_000);
assertThat(args.get("x-max-length-bytes")).isEqualTo(40_000);
assertThat(args.get("x-max-priority")).isEqualTo(8);
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");
}
@Test
public void testProducerProperties() throws Exception {
RabbitTestBinder binder = getBinder();

11
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java
View File

@@ -75,6 +75,7 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
}
this.exchanges.add(properties.getExtension().getPrefix() + name);
this.prefixes.add(properties.getExtension().getPrefix());
deadLetters(properties.getExtension());
return super.bindConsumer(name, group, moduleInputChannel, properties);
}
@@ -89,9 +90,19 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
}
}
this.prefixes.add(properties.getExtension().getPrefix());
deadLetters(properties.getExtension());
return super.bindProducer(name, moduleOutputChannel, properties);
}
private void deadLetters(RabbitCommonProperties properties) {
if (properties.getDeadLetterExchange() != null) {
this.exchanges.add(properties.getDeadLetterQueueName());
}
if (properties.getDeadLetterQueueName() != null) {
this.queues.add(properties.getDeadLetterQueueName());
}
}
@Override
public void cleanup() {
for (String queue : this.queues) {