diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 80eed476e..438a06334 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -187,6 +187,11 @@ Requires the delayed message exchange plugin on the broker. The `x-delayed-type` argument is set to the `exchangeType`. + Default: `false`. +dlqBindingArguments:: +Arguments applied when binding the dlq to the dead letter exchange; used with `headers` `deadLetterExchangeType` to specify headers to match on. +For example `...dlqBindingArguments.x-match=any`, `...dlqBindingArguments.someHeader=someValue`. ++ +Default: empty dlqDeadLetterExchange:: If a DLQ is declared, a DLX to assign to that queue. + @@ -239,7 +244,7 @@ If `declareExchange` is true, whether the exchange should be durable (that is, i + Default: `true`. exchangeType:: -The exchange type: `direct`, `fanout` or `topic` for non-partitioned destinations and `direct` or `topic` for partitioned destinations. +The exchange type: `direct`, `fanout`, `headers` or `topic` for non-partitioned destinations and `direct`, headers or `topic` for partitioned destinations. + Default: `topic`. exclusive:: @@ -309,6 +314,11 @@ prefix:: A prefix to be added to the name of the `destination` and queues. + Default: "". +queueBindingArguments:: +Arguments applied when binding the queue to the exchange; used with `headers` `exchangeType` to specify headers to match on. +For example `...queueBindingArguments.x-match=any`, `...queueBindingArguments.someHeader=someValue`. ++ +Default: empty queueDeclarationRetries:: The number of times to retry consuming from a queue if it is missing. Relevant only when `missingQueuesFatal` is `true`. @@ -463,6 +473,12 @@ deliveryMode:: The delivery mode. + Default: `PERSISTENT`. +dlqBindingArguments:: +Arguments applied when binding the dlq to the dead letter exchange; used with `headers` `deadLetterExchangeType` to specify headers to match on. +For example `...dlqBindingArguments.x-match=any`, `...dlqBindingArguments.someHeader=someValue`. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: empty dlqDeadLetterExchange:: When a DLQ is declared, a DLX to assign to that queue. Applies only if `requiredGroups` are provided and then only to those groups. @@ -513,7 +529,7 @@ If `declareExchange` is `true`, whether the exchange should be durable (survives + Default: `true`. exchangeType:: -The exchange type: `direct`, `fanout` or `topic` for non-partitioned destinations and `direct` or `topic` for partitioned destinations. +The exchange type: `direct`, `fanout`, `headers` or `topic` for non-partitioned destinations and `direct`, `headers` or `topic` for partitioned destinations. + Default: `topic`. expires:: @@ -551,6 +567,12 @@ prefix:: A prefix to be added to the name of the `destination` exchange. + Default: "". +queueBindingArguments:: +Arguments applied when binding the queue to the exchange; used with `headers` `exchangeType` to specify headers to match on. +For example `...queueBindingArguments.x-match=any`, `...queueBindingArguments.someHeader=someValue`. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: empty queueNameGroupOnly:: When `true`, consume from a queue with a name equal to the `group`. Otherwise the queue name is `destination.group`. diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java index 0c0fc6547..dd5317199 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java @@ -16,6 +16,9 @@ package org.springframework.cloud.stream.binder.rabbit.properties; +import java.util.HashMap; +import java.util.Map; + import org.hibernate.validator.constraints.Range; import org.springframework.amqp.core.ExchangeTypes; @@ -191,6 +194,18 @@ public abstract class RabbitCommonProperties { */ private String dlqOverflowBehavior; + /** + * A map of binding arguments to apply when binding the queue to the exchange. + * Useful for a headers exchange, for example. + */ + private Map queueBindingArguments = new HashMap<>(); + + /** + * A map of binding arguments to apply when binding the dlq to the exchange. + * Useful for a headers exchange, for example. + */ + private Map dlqBindingArguments = new HashMap<>(); + public String getExchangeType() { return this.exchangeType; } @@ -440,4 +455,20 @@ public abstract class RabbitCommonProperties { this.dlqOverflowBehavior = dlqOverflowBehavior; } + public Map getQueueBindingArguments() { + return this.queueBindingArguments; + } + + public void setQueueBindingArguments(Map queueBindingArguments) { + this.queueBindingArguments = queueBindingArguments; + } + + public Map getDlqBindingArguments() { + return this.dlqBindingArguments; + } + + public void setDlqBindingArguments(Map dlqBindingArguments) { + this.dlqBindingArguments = dlqBindingArguments; + } + } diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index 193b461b9..27d2e9862 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -33,6 +33,7 @@ import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Exchange; import org.springframework.amqp.core.ExchangeBuilder; import org.springframework.amqp.core.FanoutExchange; +import org.springframework.amqp.core.HeadersExchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -237,6 +238,18 @@ public class RabbitExchangeQueueProvisioner + (StringUtils.hasText(group) ? group : "default"); } + private Binding declareConsumerBindings(String name, + ExtendedConsumerProperties properties, + Exchange exchange, boolean partitioned, Queue queue) { + if (partitioned) { + return partitionedBinding(name, exchange, queue, properties.getExtension(), + properties.getInstanceIndex()); + } + else { + return notPartitionedBinding(exchange, queue, properties.getExtension()); + } + } + private Binding partitionedBinding(String destination, Exchange exchange, Queue queue, RabbitCommonProperties extendedProperties, int index) { String bindingKey = extendedProperties.getBindingRoutingKey(); @@ -244,6 +257,8 @@ public class RabbitExchangeQueueProvisioner bindingKey = destination; } bindingKey += "-" + index; + Map arguments = new HashMap<>(); + arguments.putAll(extendedProperties.getQueueBindingArguments()); if (exchange instanceof TopicExchange) { Binding binding = BindingBuilder.bind(queue).to((TopicExchange) exchange) .with(bindingKey); @@ -260,30 +275,25 @@ public class RabbitExchangeQueueProvisioner throw new ProvisioningException( "A fanout exchange is not appropriate for partitioned apps"); } + else if (exchange instanceof HeadersExchange) { + Binding binding = new Binding(queue.getName(), DestinationType.QUEUE, exchange.getName(), "", arguments); + declareBinding(queue.getName(), binding); + return binding; + } else { throw new ProvisioningException( "Cannot bind to a " + exchange.getType() + " exchange"); } } - private Binding declareConsumerBindings(String name, - ExtendedConsumerProperties properties, - Exchange exchange, boolean partitioned, Queue queue) { - if (partitioned) { - return partitionedBinding(name, exchange, queue, properties.getExtension(), - properties.getInstanceIndex()); - } - else { - return notPartitionedBinding(exchange, queue, properties.getExtension()); - } - } - private Binding notPartitionedBinding(Exchange exchange, Queue queue, RabbitCommonProperties extendedProperties) { String routingKey = extendedProperties.getBindingRoutingKey(); if (routingKey == null) { routingKey = "#"; } + Map arguments = new HashMap<>(); + arguments.putAll(extendedProperties.getQueueBindingArguments()); if (exchange instanceof TopicExchange) { Binding binding = BindingBuilder.bind(queue).to((TopicExchange) exchange) .with(routingKey); @@ -301,6 +311,11 @@ public class RabbitExchangeQueueProvisioner declareBinding(queue.getName(), binding); return binding; } + else if (exchange instanceof HeadersExchange) { + Binding binding = new Binding(queue.getName(), DestinationType.QUEUE, exchange.getName(), "", arguments); + declareBinding(queue.getName(), binding); + return binding; + } else { throw new ProvisioningException( "Cannot bind to a " + exchange.getType() + " exchange"); @@ -340,10 +355,12 @@ public class RabbitExchangeQueueProvisioner properties.getDeadLetterExchangeType()).durable(true) .build()); } + Map arguments = new HashMap<>(); + arguments.putAll(properties.getDlqBindingArguments()); Binding dlqBinding = new Binding(dlq.getName(), DestinationType.QUEUE, dlxName, properties.getDlqDeadLetterRoutingKey() == null ? routingKey : properties.getDeadLetterRoutingKey(), - null); + arguments); declareBinding(dlqName, dlqBinding); if (properties instanceof RabbitConsumerProperties && ((RabbitConsumerProperties) properties).isRepublishToDlq()) { @@ -352,7 +369,7 @@ public class RabbitExchangeQueueProvisioner * does not know about partitioning */ declareBinding(dlqName, new Binding(dlq.getName(), DestinationType.QUEUE, - dlxName, baseQueueName, null)); + dlxName, baseQueueName, arguments)); } } } 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 019c65af6..6a3dfc7b9 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 @@ -21,6 +21,7 @@ import java.io.StringWriter; import java.lang.reflect.Constructor; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -90,7 +91,6 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.Lifecycle; -import org.springframework.context.support.GenericApplicationContext; import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.integration.amqp.support.NackedAmqpMessageException; @@ -639,6 +639,56 @@ public class RabbitBinderTests extends assertThat(container.isRunning()).isFalse(); } + @Test + public void testConsumerPropertiesWithHeaderExchanges() throws Exception { + RabbitTestBinder binder = getBinder(); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.getExtension().setExchangeType(ExchangeTypes.HEADERS); + properties.getExtension().setAutoBindDlq(true); + properties.getExtension().setDeadLetterExchange(ExchangeTypes.HEADERS); + properties.getExtension().setDeadLetterExchange("propsHeader.dlx"); + Map queueBindingArguments = new HashMap<>(); + queueBindingArguments.put("x-match", "any"); + queueBindingArguments.put("foo", "bar"); + properties.getExtension().setQueueBindingArguments(queueBindingArguments); + properties.getExtension().setDlqBindingArguments(queueBindingArguments); + + String group = "bindingArgs"; + Binding consumerBinding = binder.bindConsumer("propsHeader", group, + 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(); + assertThat(container.getQueueNames()[0]).isEqualTo("propsHeader." + group); + Client client = new Client("http://guest:guest@localhost:15672/api/"); + List bindings = client.getBindingsBySource("/", "propsHeader"); + int n = 0; + while (n++ < 100 && bindings == null || bindings.size() < 1) { + Thread.sleep(100); + bindings = client.getBindingsBySource("/", "propsHeader"); + } + assertThat(bindings.size()).isEqualTo(1); + assertThat(bindings.get(0).getSource()).isEqualTo("propsHeader"); + assertThat(bindings.get(0).getDestination()).isEqualTo("propsHeader." + group); + assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("x-match", v -> assertThat(v).isEqualTo("any")); + assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("foo", v -> assertThat(v).isEqualTo("bar")); + + bindings = client.getBindingsBySource("/", "propsHeader.dlx"); + n = 0; + while (n++ < 100 && bindings == null || bindings.size() < 1) { + Thread.sleep(100); + bindings = client.getBindingsBySource("/", "propsHeader.dlx"); + } + assertThat(bindings.size()).isEqualTo(1); + assertThat(bindings.get(0).getSource()).isEqualTo("propsHeader.dlx"); + assertThat(bindings.get(0).getDestination()).isEqualTo("propsHeader." + group + ".dlq"); + assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("x-match", v -> assertThat(v).isEqualTo("any")); + assertThat(bindings.get(0).getArguments()).hasEntrySatisfying("foo", v -> assertThat(v).isEqualTo("bar")); + } + @Test public void testProducerProperties() throws Exception { RabbitTestBinder binder = getBinder(); @@ -658,9 +708,9 @@ public class RabbitBinderTests extends Boolean.class)).isFalse(); ExtendedProducerProperties producerProperties = createProducerProperties(); - ((GenericApplicationContext)this.applicationContext).registerBean("pkExtractor", + this.applicationContext.registerBean("pkExtractor", TestPartitionKeyExtractorClass.class, () -> new TestPartitionKeyExtractorClass()); - ((GenericApplicationContext)this.applicationContext).registerBean("pkSelector", + this.applicationContext.registerBean("pkSelector", TestPartitionSelectorClass.class, () -> new TestPartitionSelectorClass()); producerProperties.setPartitionKeyExtractorName("pkExtractor"); producerProperties.setPartitionSelectorName("pkSelector"); @@ -890,8 +940,8 @@ public class RabbitBinderTests extends ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.getExtension().setPrefix("bindertest."); - ((GenericApplicationContext)this.applicationContext).registerBean("pkExtractor", PartitionTestSupport.class, () -> new PartitionTestSupport()); - ((GenericApplicationContext)this.applicationContext).registerBean("pkSelector", PartitionTestSupport.class, () -> new PartitionTestSupport()); + this.applicationContext.registerBean("pkExtractor", PartitionTestSupport.class, () -> new PartitionTestSupport()); + this.applicationContext.registerBean("pkSelector", PartitionTestSupport.class, () -> new PartitionTestSupport()); producerProperties.getExtension().setAutoBindDlq(true); producerProperties.setPartitionKeyExtractorName("pkExtractor"); producerProperties.setPartitionSelectorName("pkSelector"); @@ -1008,8 +1058,8 @@ public class RabbitBinderTests extends ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.getExtension().setPrefix("bindertest."); producerProperties.getExtension().setAutoBindDlq(true); - ((GenericApplicationContext)this.applicationContext).registerBean("pkExtractor", PartitionTestSupport.class, () -> new PartitionTestSupport()); - ((GenericApplicationContext)this.applicationContext).registerBean("pkSelector", PartitionTestSupport.class, () -> new PartitionTestSupport()); + this.applicationContext.registerBean("pkExtractor", PartitionTestSupport.class, () -> new PartitionTestSupport()); + this.applicationContext.registerBean("pkSelector", PartitionTestSupport.class, () -> new PartitionTestSupport()); producerProperties.setPartitionKeyExtractorName("pkExtractor"); producerProperties.setPartitionSelectorName("pkSelector"); producerProperties.setPartitionCount(2); @@ -1130,7 +1180,7 @@ public class RabbitBinderTests extends properties.getExtension().setPrefix("bindertest."); properties.getExtension().setAutoBindDlq(true); properties.setRequiredGroups("dlqPartGrp"); - ((GenericApplicationContext)this.applicationContext).registerBean("pkExtractor", PartitionTestSupport.class, () -> new PartitionTestSupport()); + this.applicationContext.registerBean("pkExtractor", PartitionTestSupport.class, () -> new PartitionTestSupport()); properties.setPartitionKeyExtractorName("pkExtractor"); properties.setPartitionSelectorName("pkExtractor"); properties.setPartitionCount(2);