GH-34: Support User Infrastructure and Delayed Ex.

Resolves #34
Resolves #26

- Add properties to support user configuration of exchanges/queues
-- Disable binding and exchange declaration, giving complete control
-- Allow the exchange type to be specified and, for non-partitioned destinations, the routing key.
- Add support for the Delayed Message Exchange broker plugin

Doc Polishing

Polishing; add delay header; add outbound routingKey config

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

@@ -100,11 +100,33 @@ autoBindDlq::
   Whether to automatically declare the DLQ and bind it to the binder DLX.
 +
 Default: `false`.
+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`.
+declareExchange::
+  Whether to declare the exchange for the destination.
++
+Default: `true`.
+delayedExchange::
+  Whether to declare the exchange as a `Delayed Message Exchange` - requires the delayed message exchange plugin on the broker.
+  The `x-delayed-type` argument is set to the `exchangeType`.
++
+Default: `false`.
 durableSubscription::
   Whether subscription should be durable.
 Only effective if `group` is also set.
 +
 Default: `true`.
+bindingRoutingKey::
+  The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`).
+  for partitioned destinations `-<instanceIndex>` will be appended.
++
+Default: `#`.
+exchangeType::
+  The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations.
++
+Default: `topic`.
 maxConcurrency::
   Default: `1`.
 prefetch::
@@ -167,18 +189,42 @@ batchBufferLimit::
   Default: `10000`.
 batchTimeout::
   Default: `5000`.
+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.
+  Only applies if `requiredGroups` are provided and then only to those groups.
++
+Default: `true`.
 compress::
   Whether data should be compressed when sent.
 +
 Default: `false`.
-transacted::
-  Whether to use transacted channels.
+declareExchange::
+  Whether to declare the exchange for the destination.
++
+Default: `true`.
+delay::
+  A SpEL expression to evaluate the delay to apply to the message (`x-delay` header) - has no effect if the exchange is not a delayed message exchange.
++
+Default: No `x-delay` header is set.
+delayedExchange::
+  Whether to declare the exchange as a `Delayed Message Exchange` - requires the delayed message exchange plugin on the broker.
+  The `x-delayed-type` argument is set to the `exchangeType`.
 +
 Default: `false`.
 deliveryMode::
   Delivery mode.
 +
 Default: `PERSISTENT`.
+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.
++
+Default: `#`.
+exchangeType::
+  The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations.
++
+Default: `topic`.
 prefix::
   A prefix to be added to the name of the `destination` exchange.
 +
@@ -191,6 +237,14 @@ replyHeaderPatterns::
   The reply headers to be transported.
 +
 Default: `[STANDARD_REPLY_HEADERS,'*']`.
+routingKeyExpression::
+  A SpEL expression to determine the routing key to use when publishing messages.
++
+Default: `destination` or `destination-<partition>` for partitioned destinations.
+transacted::
+  Whether to use transacted channels.
++
+Default: `false`.
 
 [NOTE]
 ====

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

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.rabbit;
+
+import org.springframework.amqp.core.ExchangeTypes;
+
+/**
+ * @author Gary Russell
+ * @since 1.2
+ *
+ */
+public abstract class RabbitCommonProperties {
+
+	/**
+	 * type of exchange to declare (if necessary, and declareExchange is true).
+	 */
+	private String exchangeType = ExchangeTypes.TOPIC;
+
+	/**
+	 * whether to declare the exchange
+	 */
+	private boolean declareExchange = true;
+
+	/**
+	 * whether a delayed message exchange should be used
+	 */
+	private boolean delayedExchange = false;
+
+	/**
+	 * whether to bind a queue (or queues when partitioned) to the exchange
+	 */
+	private boolean bindQueue = true;
+
+	/**
+	 * The routing key to bind (default # for non-partitioned, destination-instanceIndex for partitioned)
+	 */
+	private String bindingRoutingKey;
+
+	public String getExchangeType() {
+		return this.exchangeType;
+	}
+
+	public void setExchangeType(String exchangeType) {
+		this.exchangeType = exchangeType;
+	}
+
+	public boolean isDeclareExchange() {
+		return this.declareExchange;
+	}
+
+	public void setDeclareExchange(boolean declareExchange) {
+		this.declareExchange = declareExchange;
+	}
+
+	public boolean isDelayedExchange() {
+		return this.delayedExchange;
+	}
+
+	public void setDelayedExchange(boolean delayedExchange) {
+		this.delayedExchange = delayedExchange;
+	}
+
+	public boolean isBindQueue() {
+		return this.bindQueue;
+	}
+
+	public void setBindQueue(boolean bindQueue) {
+		this.bindQueue = bindQueue;
+	}
+
+	public String getBindingRoutingKey() {
+		return this.bindingRoutingKey;
+	}
+
+	public void setBindingRoutingKey(String routingKey) {
+		this.bindingRoutingKey = routingKey;
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,8 +23,9 @@ import org.springframework.util.Assert;
 
 /**
  * @author Marius Bogoevici
+ * @author Gary Russell
  */
-public class RabbitConsumerProperties {
+public class RabbitConsumerProperties extends RabbitCommonProperties {
 
 	private String prefix = "";
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,17 +16,18 @@
 
 package org.springframework.cloud.stream.binder.rabbit;
 
+import java.lang.reflect.Constructor;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
-import org.aopalliance.aop.Advice;
-
 import org.springframework.amqp.AmqpConnectException;
 import org.springframework.amqp.core.AnonymousQueue;
 import org.springframework.amqp.core.BindingBuilder;
 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.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.core.Queue;
@@ -67,6 +68,7 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.retry.interceptor.RetryOperationsInterceptor;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
 import com.rabbitmq.client.AMQP;
@@ -84,7 +86,7 @@ import com.rabbitmq.client.Envelope;
  */
 public class RabbitMessageChannelBinder
 		extends AbstractMessageChannelBinder<ExtendedConsumerProperties<RabbitConsumerProperties>,
-		ExtendedProducerProperties<RabbitProducerProperties>, Queue, TopicExchange>
+		ExtendedProducerProperties<RabbitProducerProperties>, Queue, Exchange>
 		implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
 
 	private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
@@ -228,7 +230,7 @@ public class RabbitMessageChannelBinder
 					.recoverer(determineRecoverer(baseQueueName, properties.getExtension().getPrefix(),
 							properties.getExtension().isRepublishToDlq()))
 					.build();
-			listenerContainer.setAdviceChain(new Advice[] {retryInterceptor});
+			listenerContainer.setAdviceChain(retryInterceptor);
 		}
 		listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
 		listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
@@ -262,9 +264,10 @@ public class RabbitMessageChannelBinder
 		}
 		String prefix = properties.getExtension().getPrefix();
 		String exchangeName = applyPrefix(prefix, name);
-		TopicExchange exchange = new TopicExchange(exchangeName);
-		declareExchange(exchangeName, exchange);
-
+		Exchange exchange = buildExchange(properties.getExtension(), exchangeName);
+		if (properties.getExtension().isDeclareExchange()) {
+			declareExchange(exchangeName, exchange);
+		}
 		String queueName = applyPrefix(prefix, baseQueueName);
 		boolean partitioned = !anonymous && properties.isPartitioned();
 		boolean durable = !anonymous && properties.getExtension().isDurableSubscription();
@@ -288,12 +291,8 @@ public class RabbitMessageChannelBinder
 			}
 		}
 		declareQueue(queueName, queue);
-		if (partitioned) {
-			String bindingKey = String.format("%s-%d", name, properties.getInstanceIndex());
-			declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(bindingKey));
-		}
-		else {
-			declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#"));
+		if (properties.getExtension().isBindQueue()) {
+			declareConsumerBindings(name, properties, exchange, partitioned, queue);
 		}
 		if (durable) {
 			autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName,
@@ -329,16 +328,18 @@ public class RabbitMessageChannelBinder
 	}
 
 	@Override
-	protected TopicExchange createProducerDestinationIfNecessary(String name,
+	protected Exchange createProducerDestinationIfNecessary(String name,
 			ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
 		String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name);
-		TopicExchange exchange = new TopicExchange(exchangeName);
-		declareExchange(exchangeName, exchange);
+		Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName);
+		if (producerProperties.getExtension().isDeclareExchange()) {
+			declareExchange(exchangeName, exchange);
+		}
 		return exchange;
 	}
 
 	@Override
-	protected MessageHandler createProducerMessageHandler(final TopicExchange exchange,
+	protected MessageHandler createProducerMessageHandler(final Exchange exchange,
 			ExtendedProducerProperties<RabbitProducerProperties> properties)
 			throws Exception {
 		String prefix = properties.getExtension().getPrefix();
@@ -346,24 +347,38 @@ public class RabbitMessageChannelBinder
 		String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length());
 		final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(buildRabbitTemplate(properties.getExtension()));
 		endpoint.setExchangeName(exchange.getName());
+		RabbitProducerProperties extendedProperties = properties.getExtension();
+		String routingKeyExpression = extendedProperties.getRoutingKeyExpression();
 		if (!properties.isPartitioned()) {
-			endpoint.setRoutingKey(destination);
+			if (routingKeyExpression == null) {
+				endpoint.setRoutingKey(destination);
+			}
+			else {
+				endpoint.setRoutingKeyExpressionString(routingKeyExpression);
+			}
 		}
 		else {
-			endpoint.setRoutingKeyExpression(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression(
-					destination)));
+			if (routingKeyExpression == null) {
+				endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination));
+			}
+			else {
+				endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(routingKeyExpression));
+			}
+		}
+		if (extendedProperties.getDelayExpression() != null) {
+			endpoint.setDelayExpressionString(extendedProperties.getDelayExpression());
 		}
 		for (String requiredGroupName : properties.getRequiredGroups()) {
 			String baseQueueName = exchangeName + "." + requiredGroupName;
 			if (!properties.isPartitioned()) {
 				Queue queue = new Queue(baseQueueName, true, false, false,
-						queueArgs(baseQueueName, prefix, properties.getExtension().isAutoBindDlq()));
+						queueArgs(baseQueueName, prefix, extendedProperties.isAutoBindDlq()));
 				declareQueue(baseQueueName, queue);
-				autoBindDLQ(baseQueueName, baseQueueName, properties.getExtension().getPrefix(),
-						properties.getExtension().isAutoBindDlq());
-				org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(
-						destination);
-				declareBinding(baseQueueName, binding);
+				autoBindDLQ(baseQueueName, baseQueueName, extendedProperties.getPrefix(),
+						extendedProperties.isAutoBindDlq());
+				if (extendedProperties.isBindQueue()) {
+					notPartitionedBinding(exchange, queue, extendedProperties);
+				}
 			}
 			else {
 				// if the stream is partitioned, create one queue for each target partition for the default group
@@ -371,21 +386,22 @@ public class RabbitMessageChannelBinder
 					String partitionSuffix = "-" + i;
 					String partitionQueueName = baseQueueName + partitionSuffix;
 					Queue queue = new Queue(partitionQueueName, true, false, false,
-							queueArgs(partitionQueueName, properties.getExtension().getPrefix(),
-									properties.getExtension().isAutoBindDlq()));
+							queueArgs(partitionQueueName, extendedProperties.getPrefix(),
+									extendedProperties.isAutoBindDlq()));
 					declareQueue(queue.getName(), queue);
-					autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties.getExtension().getPrefix(),
-							properties.getExtension().isAutoBindDlq());
-					declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange)
-							.with(destination + partitionSuffix));
+					autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, extendedProperties.getPrefix(),
+							extendedProperties.isAutoBindDlq());
+					if (extendedProperties.isBindQueue()) {
+						partitionedBinding(destination, exchange, queue, extendedProperties, i);
+					}
 				}
 			}
 		}
 		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper();
-		mapper.setRequestHeaderNames(properties.getExtension().getRequestHeaderPatterns());
-		mapper.setReplyHeaderNames(properties.getExtension().getReplyHeaderPatterns());
+		mapper.setRequestHeaderNames(extendedProperties.getRequestHeaderPatterns());
+		mapper.setReplyHeaderNames(extendedProperties.getReplyHeaderPatterns());
 		endpoint.setHeaderMapper(mapper);
-		endpoint.setDefaultDeliveryMode(properties.getExtension().getDeliveryMode());
+		endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode());
 		endpoint.setBeanFactory(this.getBeanFactory());
 		endpoint.afterPropertiesSet();
 		return endpoint;
@@ -452,6 +468,22 @@ public class RabbitMessageChannelBinder
 		addToAutoDeclareContext(beanName, queue);
 	}
 
+	private Exchange buildExchange(RabbitCommonProperties properties, String exchangeName) {
+		try {
+			// TODO Make the ctor public in Spring-AMQP - AMQP-695
+			Constructor<ExchangeBuilder> ctor = ExchangeBuilder.class.getDeclaredConstructor(String.class, String.class);
+			ReflectionUtils.makeAccessible(ctor);
+			ExchangeBuilder builder = ctor.newInstance(exchangeName, properties.getExchangeType());
+			if (properties.isDelayedExchange()) {
+				builder.delayed();
+			}
+			return builder.build();
+		}
+		catch (Exception e) {
+			throw new IllegalStateException("Failed to create exchange object", e);
+		}
+	}
+
 	private void declareExchange(final String rootName, final Exchange exchange) {
 		try {
 			this.rabbitAdmin.declareExchange(exchange);
@@ -465,6 +497,65 @@ public class RabbitMessageChannelBinder
 		addToAutoDeclareContext(rootName + ".exchange", exchange);
 	}
 
+	private void declareConsumerBindings(String name, ExtendedConsumerProperties<RabbitConsumerProperties> properties,
+			Exchange exchange, boolean partitioned, Queue queue) {
+		if (partitioned) {
+			partitionedBinding(name, exchange, queue, properties.getExtension(), properties.getInstanceIndex());
+		}
+		else {
+			notPartitionedBinding(exchange, queue, properties.getExtension());
+		}
+	}
+
+	private void partitionedBinding(String destination, Exchange exchange, Queue queue,
+			RabbitCommonProperties extendedProperties, int index) {
+		String bindingKey = extendedProperties.getBindingRoutingKey();
+		if (bindingKey == null) {
+			bindingKey = destination;
+		}
+		bindingKey += "-" + index;
+		if (exchange instanceof TopicExchange) {
+			declareBinding(queue.getName(), BindingBuilder.bind(queue)
+								.to((TopicExchange) exchange)
+								.with(bindingKey));
+		}
+		else if (exchange instanceof DirectExchange) {
+			declareBinding(queue.getName(), BindingBuilder.bind(queue)
+					.to((DirectExchange) exchange)
+					.with(bindingKey));
+		}
+		else if (exchange instanceof FanoutExchange) {
+			throw new IllegalStateException("A fanout exchange is not appropriate for partitioned apps");
+		}
+		else {
+			throw new IllegalStateException("Cannot bind to a " + exchange.getType() + " exchange");
+		}
+	}
+
+	private void notPartitionedBinding(Exchange exchange, Queue queue, RabbitCommonProperties extendedProperties) {
+		String routingKey = extendedProperties.getBindingRoutingKey();
+		if (routingKey == null) {
+			routingKey = "#";
+		}
+		if (exchange instanceof TopicExchange) {
+			declareBinding(queue.getName(), BindingBuilder.bind(queue)
+								.to((TopicExchange) exchange)
+								.with(routingKey));
+		}
+		else if (exchange instanceof DirectExchange) {
+			declareBinding(queue.getName(), BindingBuilder.bind(queue)
+					.to((DirectExchange) exchange)
+					.with(routingKey));
+		}
+		else if (exchange instanceof FanoutExchange) {
+			declareBinding(queue.getName(), BindingBuilder.bind(queue)
+					.to((FanoutExchange) exchange));
+		}
+		else {
+			throw new IllegalStateException("Cannot bind to a " + exchange.getType() + " exchange");
+		}
+	}
+
 	private void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) {
 		try {
 			this.rabbitAdmin.declareBinding(binding);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,7 +24,7 @@ import org.springframework.amqp.core.MessageDeliveryMode;
  * @author Marius Bogoevici
  * @author Gary Russell
  */
-public class RabbitProducerProperties {
+public class RabbitProducerProperties extends RabbitCommonProperties {
 
 	private String prefix = "";
 
@@ -48,6 +48,16 @@ public class RabbitProducerProperties {
 
 	private String[] replyHeaderPatterns = new String[] {"STANDARD_REPLY_HEADERS", "*"};
 
+	/**
+	 * When using a delayed message exchange, a SpEL expression to determine the delay to apply to messages
+	 */
+	private String delayExpression;
+
+	/**
+	 * A custom routing key when publishing messages; default is the destination name; suffixed by "-partition" when partitioned
+	 */
+	private String routingKeyExpression;
+
 	public String getPrefix() {
 		return prefix;
 	}
@@ -139,4 +149,20 @@ public class RabbitProducerProperties {
 		this.transacted = transacted;
 	}
 
+	public String getDelayExpression() {
+		return this.delayExpression;
+	}
+
+	public void setDelayExpression(String delayExpression) {
+		this.delayExpression = delayExpression;
+	}
+
+	public String getRoutingKeyExpression() {
+		return this.routingKeyExpression;
+	}
+
+	public void setRoutingKeyExpression(String routingKeyExpression) {
+		this.routingKeyExpression = routingKeyExpression;
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -35,10 +35,15 @@ import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 
 import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.ExchangeTypes;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitManagementTemplate;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.support.AmqpHeaders;
@@ -82,7 +87,7 @@ public class RabbitBinderTests extends
 	public static final String TEST_PREFIX = "bindertest.";
 
 	@Rule
-	public RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport();
+	public RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true);
 
 	@Override
 	protected RabbitTestBinder getBinder() {
@@ -185,13 +190,81 @@ public class RabbitBinderTests extends
 		assertThat(endpoint.isRunning()).isFalse();
 	}
 
+	@Test
+	public void testConsumerPropertiesWithUserInfrastructureNoBind() throws Exception {
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		Queue queue = new Queue("propsUser1.infra");
+		admin.declareQueue(queue);
+		DirectExchange exchange = new DirectExchange("propsUser1");
+		admin.declareExchange(exchange);
+		admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("foo"));
+
+		RabbitTestBinder binder = getBinder();
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setDeclareExchange(false);
+		properties.getExtension().setBindQueue(false);
+
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("propsUser1", "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("/", exchange.getName());
+		assertThat(bindings.size()).isEqualTo(1);
+	}
+
+	@Test
+	public void testConsumerPropertiesWithUserInfrastructureCustomExchangeAndRK() throws Exception {
+		RabbitTestBinder binder = getBinder();
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		properties.getExtension().setExchangeType(ExchangeTypes.DIRECT);
+		properties.getExtension().setBindingRoutingKey("foo");
+//		properties.getExtension().setDelayedExchange(true); // requires delayed message exchange plugin; tested locally
+
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("propsUser2", "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("/", "propsUser2");
+		int n = 0;
+		while (n++ < 100 && bindings == null || bindings.size() < 1) {
+			Thread.sleep(100);
+			bindings = rmt.getBindingsForExchange("/", "propsUser2");
+		}
+		assertThat(bindings.size()).isEqualTo(1);
+		assertThat(bindings.get(0).getExchange()).isEqualTo("propsUser2");
+		assertThat(bindings.get(0).getDestination()).isEqualTo("propsUser2.infra");
+		assertThat(bindings.get(0).getRoutingKey()).isEqualTo("foo");
+
+//		// TODO: AMQP-696
+//		// Exchange exchange = rmt.getExchange("propsUser2");
+//		ExchangeInfo ei = rmt.getClient().getExchange("/", "propsUser2"); // requires delayed message exchange plugin
+//		assertThat(ei.getType()).isEqualTo("x-delayed-message");
+//		assertThat(ei.getArguments().get("x-delayed-type")).isEqualTo("direct");
+
+		Exchange exchange = rmt.getExchange("propsUser2");
+		while (n++ < 100 && exchange == null) {
+			Thread.sleep(100);
+			exchange = rmt.getExchange("propsUser2");
+		}
+		assertThat(exchange).isInstanceOf(DirectExchange.class);
+	}
+
 	@Test
 	public void testProducerProperties() throws Exception {
 		RabbitTestBinder binder = getBinder();
 		Binding<MessageChannel> producerBinding = binder.bindProducer("props.0",
 				createBindableChannel("input", new BindingProperties()),
 				createProducerProperties());
-		@SuppressWarnings("unchecked")
 		Lifecycle endpoint = extractEndpoint(producerBinding);
 		MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode",
 				MessageDeliveryMode.class);
@@ -214,18 +287,28 @@ public class RabbitBinderTests extends
 		producerProperties.setPartitionSelectorClass(TestPartitionSelectorClass.class);
 		producerProperties.setPartitionCount(1);
 		producerProperties.getExtension().setTransacted(true);
+		producerProperties.getExtension().setDelayExpression("42");
+		producerProperties.setRequiredGroups("prodPropsRequired");
 
 		BindingProperties producerBindingProperties = createProducerBindingProperties(producerProperties);
-		producerBinding = binder.bindProducer("props.0", createBindableChannel("output", producerBindingProperties),
+		DirectChannel channel = createBindableChannel("output", producerBindingProperties);
+		producerBinding = binder.bindProducer("props.0", channel,
 				producerProperties);
 		endpoint = extractEndpoint(producerBinding);
 		assertThat(TestUtils.getPropertyValue(endpoint, "routingKeyExpression", SpelExpression.class)
 				.getExpressionString()).isEqualTo("'props.0-' + headers['partition']");
+		assertThat(TestUtils.getPropertyValue(endpoint, "delayExpression", SpelExpression.class)
+				.getExpressionString()).isEqualTo("42");
 		mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode", MessageDeliveryMode.class);
 		assertThat(mode).isEqualTo(MessageDeliveryMode.NON_PERSISTENT);
 		assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional", Boolean.class))
 				.isTrue();
 		verifyFooRequestProducer(endpoint);
+		channel.send(new GenericMessage<>("foo"));
+		org.springframework.amqp.core.Message received = new RabbitTemplate(this.rabbitAvailableRule.getResource())
+				.receive("foo.props.0.prodPropsRequired-0", 10_000);
+		assertThat(received).isNotNull();
+		assertThat(received.getMessageProperties().getReceivedDelay()).isEqualTo(42);
 
 		producerBinding.unbind();
 		assertThat(endpoint.isRunning()).isFalse();
@@ -663,7 +746,6 @@ public class RabbitBinderTests extends
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testBatchingAndCompression() throws Exception {
-		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
 		RabbitTestBinder binder = getBinder();
 		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
@@ -910,20 +992,22 @@ public class RabbitBinderTests extends
 		};
 	}
 
-	private static class TestPartitionKeyExtractorClass implements PartitionKeyExtractorStrategy {
+	public static class TestPartitionKeyExtractorClass implements PartitionKeyExtractorStrategy {
 
 		@Override
 		public Object extractKey(Message<?> message) {
-			return null;
+			return 0;
 		}
+
 	}
 
-	private static class TestPartitionSelectorClass implements PartitionSelectorStrategy {
+	public static class TestPartitionSelectorClass implements PartitionSelectorStrategy {
 
 		@Override
 		public int selectPartition(Object key, int partitionCount) {
 			return 0;
 		}
+
 	}
 
 }

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

@@ -74,6 +74,7 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
 			this.queues.add(properties.getExtension().getPrefix() + name + ("." + group));
 		}
 		this.exchanges.add(properties.getExtension().getPrefix() + name);
+		this.prefixes.add(properties.getExtension().getPrefix());
 		return super.bindConsumer(name, group, moduleInputChannel, properties);
 	}
 
@@ -82,6 +83,12 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
 			ExtendedProducerProperties<RabbitProducerProperties> properties) {
 		this.queues.add(properties.getExtension().getPrefix() + name + ".default");
 		this.exchanges.add(properties.getExtension().getPrefix() + name);
+		if (properties.getRequiredGroups() != null) {
+			for (String group : properties.getRequiredGroups()) {
+				this.queues.add(properties.getExtension().getPrefix() + name + "." + group);
+			}
+		}
+		this.prefixes.add(properties.getExtension().getPrefix());
 		return super.bindProducer(name, moduleOutputChannel, properties);
 	}