GH-254: Add support for headers exchange

Resolves #254

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`.

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<String, String>  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<String, String>  dlqBindingArguments = new HashMap<>();
+
 	public String getExchangeType() {
 		return this.exchangeType;
 	}
@@ -440,4 +455,20 @@ public abstract class RabbitCommonProperties {
 		this.dlqOverflowBehavior = dlqOverflowBehavior;
 	}
 
+	public Map<String, String> getQueueBindingArguments() {
+		return this.queueBindingArguments;
+	}
+
+	public void setQueueBindingArguments(Map<String, String> queueBindingArguments) {
+		this.queueBindingArguments = queueBindingArguments;
+	}
+
+	public Map<String, String> getDlqBindingArguments() {
+		return this.dlqBindingArguments;
+	}
+
+	public void setDlqBindingArguments(Map<String, String> dlqBindingArguments) {
+		this.dlqBindingArguments = dlqBindingArguments;
+	}
+
 }

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<RabbitConsumerProperties> 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<String, Object> 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<RabbitConsumerProperties> 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<String, Object> 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<String, Object> 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));
 			}
 		}
 	}

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<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setExchangeType(ExchangeTypes.HEADERS);
+		properties.getExtension().setAutoBindDlq(true);
+		properties.getExtension().setDeadLetterExchange(ExchangeTypes.HEADERS);
+		properties.getExtension().setDeadLetterExchange("propsHeader.dlx");
+		Map<String, String> queueBindingArguments = new HashMap<>();
+		queueBindingArguments.put("x-match", "any");
+		queueBindingArguments.put("foo", "bar");
+		properties.getExtension().setQueueBindingArguments(queueBindingArguments);
+		properties.getExtension().setDlqBindingArguments(queueBindingArguments);
+
+		String group = "bindingArgs";
+		Binding<MessageChannel> 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<BindingInfo> 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<RabbitProducerProperties> 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<RabbitProducerProperties> 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<RabbitProducerProperties> 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);