diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index 4d9dacdeb..5db77f739 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc @@ -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 <> 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`. diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java index a2426d511..0dabe1085 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java @@ -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; + } + } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java index b786cb583..f6d122ca5 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java @@ -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; } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index b8f1518f3..5b1cf983a 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -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 queueArgs(String queueName, String prefix, boolean bindDlq) { - Map 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 queueArgs(String queueName, RabbitCommonProperties properties, boolean isDlq) { + Map 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 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) { diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java index 25eb37457..d1d98aada 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java @@ -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; } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index a5368d861..bed837197 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -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 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 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 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 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(); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 4aa869af4..3df34fca3 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -75,6 +75,7 @@ public class RabbitTestBinder extends AbstractTestBinder