Provisioning Changes

- Separate Rabbit exchange/queue provisioning from the binder using the SPI provided in core
 - Refactor the common entities needed for Rabbit into a new core module
 - Refactor binder code to reflect the SPI changes

Fixes #40

Addressing PR review comments

Addressing PR review comments

cleanup - Addressing PR review

Using ProvisioningException

Addressing PR comments

pom.xml

@@ -51,6 +51,7 @@
 		</dependencies>
 	</dependencyManagement>
 	<modules>
+		<module>spring-cloud-stream-binder-rabbit-core</module>
 		<module>spring-cloud-stream-binder-rabbit</module>
 		<module>spring-cloud-starter-stream-rabbit</module>
 		<module>spring-cloud-stream-binder-rabbit-test-support</module>

spring-cloud-stream-binder-rabbit-core/pom.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-stream-binder-rabbit-core</artifactId>
+	<packaging>jar</packaging>
+	<name>spring-cloud-stream-binder-rabbit-core</name>
+	<description>RabbitMQ binder core</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
+		<version>1.2.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-amqp</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-amqp</artifactId>
+		</dependency>
+	</dependencies>
+</project>

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitAdminException.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.admin;
 
 
 /**

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.admin;
 
 import java.net.URI;
 import java.util.ArrayList;

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.admin;
 
 import java.net.URI;
 import java.net.URISyntaxException;

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit.config;
+package org.springframework.cloud.stream.binder.rabbit.properties;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
@@ -22,7 +22,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
  * @author David Turanski
  */
 @ConfigurationProperties(prefix = "spring.cloud.stream.rabbit.binder")
-class RabbitBinderConfigurationProperties {
+public class RabbitBinderConfigurationProperties {
 
 	private String[] adminAddresses = new String[0];
 

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBindingProperties.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.properties;
 
 /**
  * @author Marius Bogoevici

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.properties;
 
 import org.hibernate.validator.constraints.Range;
 
@@ -27,6 +27,8 @@ import org.springframework.amqp.core.ExchangeTypes;
  */
 public abstract class RabbitCommonProperties {
 
+	public static final String DEAD_LETTER_EXCHANGE = "DLX";
+
 	/**
 	 * type of exchange to declare (if necessary, and declareExchange is true).
 	 */

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.properties;
 
 import javax.validation.constraints.Min;
 

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.properties;
 
 import java.util.HashMap;
 import java.util.Map;

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.rabbit;
+package org.springframework.cloud.stream.binder.rabbit.properties;
 
 import javax.validation.constraints.Min;
 

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java

@@ -0,0 +1,510 @@
+/*
+ * 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.
+ * 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.provisioning;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+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.DirectExchange;
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.ExchangeBuilder;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.cloud.stream.provisioning.ProvisioningException;
+import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * AMQP implementation for {@link ProvisioningProvider}
+ *
+ * @author Soby Chacko
+ */
+public class RabbitExchangeQueueProvisioner implements ProvisioningProvider<ExtendedConsumerProperties<RabbitConsumerProperties>,
+		ExtendedProducerProperties<RabbitProducerProperties>> {
+
+	private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
+			= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
+
+	/**
+	 * The delimiter between a group and index when constructing a binder
+	 * consumer/producer.
+	 */
+	private static final String GROUP_INDEX_DELIMITER = ".";
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	private final RabbitAdmin rabbitAdmin;
+
+	private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext();
+
+	public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory) {
+		this.rabbitAdmin = new RabbitAdmin(connectionFactory);
+		this.autoDeclareContext.refresh();
+		this.rabbitAdmin.setApplicationContext(this.autoDeclareContext);
+		this.rabbitAdmin.setIgnoreDeclarationExceptions(true);
+		this.rabbitAdmin.afterPropertiesSet();
+	}
+
+	@Override
+	public ProducerDestination provisionProducerDestination(String name,
+															ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
+		final String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name);
+		Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName);
+		if (producerProperties.getExtension().isDeclareExchange()) {
+			declareExchange(exchangeName, exchange);
+		}
+		Binding binding = null;
+		for (String requiredGroupName : producerProperties.getRequiredGroups()) {
+			String baseQueueName = exchangeName + "." + requiredGroupName;
+			if (!producerProperties.isPartitioned()) {
+				Queue queue = new Queue(baseQueueName, true, false, false,
+						queueArgs(baseQueueName, producerProperties.getExtension(), false));
+				declareQueue(baseQueueName, queue);
+				autoBindDLQ(baseQueueName, baseQueueName, producerProperties.getExtension());
+				if (producerProperties.getExtension().isBindQueue()) {
+					binding = notPartitionedBinding(exchange, queue, producerProperties.getExtension());
+				}
+			}
+			else {
+				// if the stream is partitioned, create one queue for each target partition for the default group
+				for (int i = 0; i < producerProperties.getPartitionCount(); i++) {
+					String partitionSuffix = "-" + i;
+					String partitionQueueName = baseQueueName + partitionSuffix;
+					Queue queue = new Queue(partitionQueueName, true, false, false,
+							queueArgs(partitionQueueName, producerProperties.getExtension(), false));
+					declareQueue(queue.getName(), queue);
+					autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, producerProperties.getExtension());
+					if (producerProperties.getExtension().isBindQueue()) {
+						String prefix = producerProperties.getExtension().getPrefix();
+						String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length());
+						binding = partitionedBinding(destination, exchange, queue, producerProperties.getExtension(), i);
+					}
+				}
+			}
+		}
+		return new RabbitProducerDestination(exchange, binding);
+	}
+
+	@Override
+	public ConsumerDestination provisionConsumerDestination(String name, String group, ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
+		boolean anonymous = !StringUtils.hasText(group);
+		String baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
+				: groupedName(name, group);
+		if (this.logger.isInfoEnabled()) {
+			this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name);
+		}
+		String prefix = properties.getExtension().getPrefix();
+		final String exchangeName = applyPrefix(prefix, name);
+		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();
+		Queue queue;
+		if (anonymous) {
+			queue = new Queue(queueName, false, true, true, queueArgs(queueName, properties.getExtension(), false));
+		}
+		else {
+			if (partitioned) {
+				String partitionSuffix = "-" + properties.getInstanceIndex();
+				queueName += partitionSuffix;
+			}
+			if (durable) {
+				queue = new Queue(queueName, true, false, false,
+						queueArgs(queueName, properties.getExtension(), false));
+			}
+			else {
+				queue = new Queue(queueName, false, false, true,
+						queueArgs(queueName, properties.getExtension(), false));
+			}
+		}
+		declareQueue(queueName, queue);
+		Binding binding = null;
+		if (properties.getExtension().isBindQueue()) {
+			binding = declareConsumerBindings(name, properties, exchange, partitioned, queue);
+		}
+		if (durable) {
+			autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName,
+					properties.getExtension());
+		}
+		return new RabbitConsumerDestination(queue, binding);
+	}
+
+	/**
+	 * Construct a name comprised of the name and group.
+	 *
+	 * @param name  the name.
+	 * @param group the group.
+	 * @return the constructed name.
+	 */
+	protected final String groupedName(String name, String group) {
+		return name + GROUP_INDEX_DELIMITER + (StringUtils.hasText(group) ? group : "default");
+	}
+
+	private Binding 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) {
+			Binding binding = BindingBuilder.bind(queue)
+					.to((TopicExchange) exchange)
+					.with(bindingKey);
+			declareBinding(queue.getName(), binding);
+			return binding;
+		}
+		else if (exchange instanceof DirectExchange) {
+			Binding binding = BindingBuilder.bind(queue)
+					.to((DirectExchange) exchange)
+					.with(bindingKey);
+			declareBinding(queue.getName(), binding);
+			return binding;
+		}
+		else if (exchange instanceof FanoutExchange) {
+			throw new ProvisioningException("A fanout exchange is not appropriate for partitioned apps");
+		}
+		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 = "#";
+		}
+		if (exchange instanceof TopicExchange) {
+			Binding binding = BindingBuilder.bind(queue)
+					.to((TopicExchange) exchange)
+					.with(routingKey);
+			declareBinding(queue.getName(), binding);
+			return binding;
+		}
+		else if (exchange instanceof DirectExchange) {
+			Binding binding = BindingBuilder.bind(queue)
+					.to((DirectExchange) exchange)
+					.with(routingKey);
+			declareBinding(queue.getName(), binding);
+			return binding;
+		}
+		else if (exchange instanceof FanoutExchange) {
+			Binding binding = BindingBuilder.bind(queue)
+					.to((FanoutExchange) exchange);
+			declareBinding(queue.getName(), binding);
+			return binding;
+		}
+		else {
+			throw new ProvisioningException("Cannot bind to a " + exchange.getType() + " exchange");
+		}
+	}
+
+	/**
+	 * If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original
+	 * 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 properties the properties.
+	 */
+	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;
+			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);
+			String dlxName = deadLetterExchangeName(properties);
+			final DirectExchange dlx = new DirectExchange(dlxName);
+			declareExchange(dlxName, dlx);
+			BindingBuilder.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));
+			}
+		}
+	}
+
+	/**
+	 * For binder implementations that support dead lettering, construct the name of the
+	 * dead letter entity for the underlying pipe name.
+	 *
+	 * @param name the name.
+	 */
+	public static String constructDLQName(String name) {
+		return name + ".dlq";
+	}
+
+
+	private String deadLetterExchangeName(RabbitCommonProperties properties) {
+		if (properties.getDeadLetterExchange() == null) {
+			return properties.getPrefix() + RabbitCommonProperties.DEAD_LETTER_EXCHANGE;
+		}
+		else {
+			return properties.getDeadLetterExchange();
+		}
+	}
+
+
+	private void declareQueue(String beanName, Queue queue) {
+		try {
+			this.rabbitAdmin.declareQueue(queue);
+		}
+		catch (AmqpConnectException e) {
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
+			}
+		}
+		addToAutoDeclareContext(beanName, queue);
+	}
+
+	private Map<String, Object> queueArgs(String queueName, RabbitCommonProperties properties, boolean isDlq) {
+		Map<String, Object> args = new HashMap<>();
+		if (!isDlq) {
+			if (properties.isAutoBindDlq()) {
+				String dlx;
+				if (properties.getDeadLetterExchange() != null) {
+					dlx = properties.getDeadLetterExchange();
+				}
+				else {
+					dlx = applyPrefix(properties.getPrefix(), "DLX");
+				}
+				args.put("x-dead-letter-exchange", dlx);
+				String dlRk;
+				if (properties.getDeadLetterRoutingKey() != null) {
+					dlRk = properties.getDeadLetterRoutingKey();
+				}
+				else {
+					dlRk = queueName;
+				}
+				args.put("x-dead-letter-routing-key", dlRk);
+			}
+			additionalArgs(args, properties.getExpires(), properties.getMaxLength(), properties.getMaxLengthBytes(),
+					properties.getMaxPriority(), properties.getTtl());
+		}
+		else {
+			if (properties.getDlqDeadLetterExchange() != null) {
+				args.put("x-dead-letter-exchange", properties.getDlqDeadLetterExchange());
+			}
+			if (properties.getDlqDeadLetterRoutingKey() != null) {
+				args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey());
+			}
+			additionalArgs(args, properties.getDlqExpires(), properties.getDlqMaxLength(),
+					properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl());
+		}
+		return args;
+	}
+
+	private void additionalArgs(Map<String, Object> args, Integer expires, Integer maxLength, Integer maxLengthBytes,
+								Integer maxPriority, Integer ttl) {
+		if (expires != null) {
+			args.put("x-expires", expires);
+		}
+		if (maxLength != null) {
+			args.put("x-max-length", maxLength);
+		}
+		if (maxLengthBytes != null) {
+			args.put("x-max-length-bytes", maxLengthBytes);
+		}
+		if (maxPriority != null) {
+			args.put("x-max-priority", maxPriority);
+		}
+		if (ttl != null) {
+			args.put("x-message-ttl", ttl);
+		}
+	}
+
+
+	public static String applyPrefix(String prefix, String name) {
+		return prefix + name;
+	}
+
+	private Exchange buildExchange(RabbitCommonProperties properties, String exchangeName) {
+		try {
+			ExchangeBuilder builder = new ExchangeBuilder(exchangeName, properties.getExchangeType());
+			if (properties.isDelayedExchange()) {
+				builder.delayed();
+			}
+			return builder.build();
+		}
+		catch (Exception e) {
+			throw new ProvisioningException("Failed to create exchange object", e);
+		}
+	}
+
+	private void declareExchange(final String rootName, final Exchange exchange) {
+		try {
+			this.rabbitAdmin.declareExchange(exchange);
+		}
+		catch (AmqpConnectException e) {
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug(
+						"Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
+			}
+		}
+		addToAutoDeclareContext(rootName + ".exchange", exchange);
+	}
+
+	private void addToAutoDeclareContext(String name, Object bean) {
+		synchronized (this.autoDeclareContext) {
+			if (!this.autoDeclareContext.containsBean(name)) {
+				this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean);
+			}
+		}
+	}
+
+	private void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) {
+		try {
+			this.rabbitAdmin.declareBinding(binding);
+		}
+		catch (AmqpConnectException e) {
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug(
+						"Declaration of binding: " + rootName + ".binding deferred - connection not available");
+			}
+		}
+		addToAutoDeclareContext(rootName + ".binding", binding);
+	}
+
+	public void cleanAutoDeclareContext(String name) {
+		synchronized (this.autoDeclareContext) {
+			removeSingleton(name + ".binding");
+			removeSingleton(name);
+			String dlq = name + ".dlq";
+			removeSingleton(dlq + ".binding");
+			removeSingleton(dlq);
+		}
+	}
+
+	private void removeSingleton(String name) {
+		if (this.autoDeclareContext.containsBean(name)) {
+			ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory();
+			if (beanFactory instanceof DefaultListableBeanFactory) {
+				((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
+			}
+		}
+	}
+
+	private final class RabbitProducerDestination implements ProducerDestination {
+
+		private final Exchange exchange;
+		private final Binding binding;
+
+		private RabbitProducerDestination(Exchange exchange, Binding binding) {
+			Assert.notNull(exchange, "exchange must not be null");
+			this.exchange = exchange;
+			this.binding = binding;
+		}
+
+		@Override
+		public String getName() {
+			return this.exchange.getName();
+		}
+
+		@Override
+		public String getNameForPartition(int partition) {
+			return this.exchange.getName();
+		}
+
+		@Override
+		public String toString() {
+			return "RabbitProducerDestination{" +
+					"exchange=" + exchange +
+					", binding=" + binding +
+					'}';
+		}
+	}
+
+	private final class RabbitConsumerDestination implements ConsumerDestination {
+
+		private final Queue queue;
+		private final Binding binding;
+
+		private RabbitConsumerDestination(Queue queue, Binding binding) {
+			Assert.notNull(queue, "queue must not be null");
+			this.queue = queue;
+			this.binding = binding;
+		}
+
+		@Override
+		public String toString() {
+			return "RabbitConsumerDestination{" +
+					"queue=" + queue +
+					", binding=" + binding +
+					'}';
+		}
+
+		@Override
+		public String getName() {
+			return this.queue.getName();
+		}
+	}
+}

spring-cloud-stream-binder-rabbit/pom.xml

@@ -14,6 +14,11 @@
 	</parent>
 
 	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-rabbit-core</artifactId>
+			<version>${project.version}</version>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-configuration-processor</artifactId>

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

@@ -16,29 +16,17 @@
 
 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.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;
-import org.springframework.amqp.core.FanoutExchange;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Envelope;
+
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.core.Queue;
-import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
 import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
@@ -50,14 +38,18 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
 import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
-import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
-import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
-import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
+import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
@@ -70,12 +62,8 @@ 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;
-import com.rabbitmq.client.Envelope;
-
 /**
  * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ.
  * @author Mark Fisher
@@ -88,14 +76,9 @@ import com.rabbitmq.client.Envelope;
  */
 public class RabbitMessageChannelBinder
 		extends AbstractMessageChannelBinder<ExtendedConsumerProperties<RabbitConsumerProperties>,
-		ExtendedProducerProperties<RabbitProducerProperties>, Queue, Exchange>
+		ExtendedProducerProperties<RabbitProducerProperties>>
 		implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
 
-	private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
-			= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
-
-	private static final String DEAD_LETTER_EXCHANGE = "DLX";
-
 	private static final MessagePropertiesConverter inboundMessagePropertiesConverter =
 			new DefaultMessagePropertiesConverter() {
 
@@ -108,10 +91,6 @@ public class RabbitMessageChannelBinder
 				}
 			};
 
-	private final RabbitAdmin rabbitAdmin;
-
-	private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext();
-
 	private final RabbitProperties rabbitProperties;
 
 	private ConnectionFactory connectionFactory;
@@ -128,17 +107,16 @@ public class RabbitMessageChannelBinder
 
 	private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties();
 
-	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
-		super(true, new String[0]);
+	RabbitExchangeQueueProvisioner provisioningProvider;
+
+	public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties,
+										RabbitExchangeQueueProvisioner provisioningProvider) {
+		super(true, new String[0], provisioningProvider);
 		Assert.notNull(connectionFactory, "connectionFactory must not be null");
 		Assert.notNull(rabbitProperties, "rabbitProperties must not be null");
 		this.connectionFactory = connectionFactory;
 		this.rabbitProperties = rabbitProperties;
-		this.rabbitAdmin = new RabbitAdmin(connectionFactory);
-		this.autoDeclareContext.refresh();
-		this.rabbitAdmin.setApplicationContext(this.autoDeclareContext);
-		this.rabbitAdmin.setIgnoreDeclarationExceptions(true);
-		this.rabbitAdmin.afterPropertiesSet();
+		this.provisioningProvider = provisioningProvider;
 	}
 
 	/**
@@ -194,21 +172,65 @@ public class RabbitMessageChannelBinder
 		return this.extendedBindingProperties.getExtendedConsumerProperties(channelName);
 	}
 
-
 	@Override
 	public RabbitProducerProperties getExtendedProducerProperties(String channelName) {
 		return this.extendedBindingProperties.getExtendedProducerProperties(channelName);
 	}
 
 	@Override
-	protected MessageProducer createConsumerEndpoint(String name, String group, Queue destination,
-			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
+	protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination,
+														ExtendedProducerProperties<RabbitProducerProperties> producerProperties)
+			throws Exception {
+		String prefix = producerProperties.getExtension().getPrefix();
+		String exchangeName = producerDestination.getName();
+		String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length());
+		final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(buildRabbitTemplate(producerProperties.getExtension()));
+		endpoint.setExchangeName(producerDestination.getName());
+		RabbitProducerProperties extendedProperties = producerProperties.getExtension();
+		String routingKeyExpression = extendedProperties.getRoutingKeyExpression();
+		if (!producerProperties.isPartitioned()) {
+			if (routingKeyExpression == null) {
+				endpoint.setRoutingKey(destination);
+			}
+			else {
+				endpoint.setRoutingKeyExpressionString(routingKeyExpression);
+			}
+		}
+		else {
+			if (routingKeyExpression == null) {
+				endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination));
+			}
+			else {
+				endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(routingKeyExpression));
+			}
+		}
+		if (extendedProperties.getDelayExpression() != null) {
+			endpoint.setDelayExpressionString(extendedProperties.getDelayExpression());
+		}
+		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper();
+		mapper.setRequestHeaderNames(extendedProperties.getRequestHeaderPatterns());
+		mapper.setReplyHeaderNames(extendedProperties.getReplyHeaderPatterns());
+		endpoint.setHeaderMapper(mapper);
+		endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode());
+		endpoint.setBeanFactory(this.getBeanFactory());
+		endpoint.afterPropertiesSet();
+		return endpoint;
+	}
+
+	@Override
+	protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group,
+													ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
 
 		DirectChannel convertingBridgeChannel = new DirectChannel();
 		convertingBridgeChannel.setBeanFactory(this.getBeanFactory());
-		final String baseQueueName = baseQueueName(name, group);
-		convertingBridgeChannel.setBeanName(baseQueueName + ".bridge");
 
+		String prefix = properties.getExtension().getPrefix();
+		String destination = consumerDestination.getName();
+		String prefixStripped = (StringUtils.isEmpty(prefix) || !destination.startsWith(prefix)) ? destination
+				: destination.substring(prefix.length());
+		String baseQueueName = StringUtils.hasText(group) ? prefixStripped.substring(0, prefixStripped.indexOf(group)) + group : prefixStripped;
+
+		convertingBridgeChannel.setBeanName(baseQueueName + ".bridge");
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
 				this.connectionFactory);
 		listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
@@ -224,8 +246,8 @@ public class RabbitMessageChannelBinder
 		listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
 		listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
 		listenerContainer.setTxSize(properties.getExtension().getTxSize());
-		listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(destination.getName() + "-"));
-		listenerContainer.setQueues(destination);
+		listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
+		listenerContainer.setQueueNames(consumerDestination.getName());
 		if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) {
 			RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
 					.retryOperations(buildRetryTemplate(properties))
@@ -249,63 +271,19 @@ public class RabbitMessageChannelBinder
 		return adapter;
 	}
 
-	@Override
-	protected void afterUnbindConsumer(String name, String group,
-			ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
-		cleanAutoDeclareContext(consumerProperties.getExtension().getPrefix(), baseQueueName(name, group));
-	}
-
-	@Override
-	protected Queue createConsumerDestinationIfNecessary(String name, String group,
-			ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
-		boolean anonymous = !StringUtils.hasText(group);
-		String baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
-				: groupedName(name, group);
-		if (this.logger.isInfoEnabled()) {
-			this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name);
-		}
-		String prefix = properties.getExtension().getPrefix();
-		String exchangeName = applyPrefix(prefix, name);
-		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();
-		Queue queue;
-
-		if (anonymous) {
-			queue = new Queue(queueName, false, true, true, queueArgs(queueName, properties.getExtension(), false));
+	private String deadLetterExchangeName(RabbitCommonProperties properties) {
+		if (properties.getDeadLetterExchange() == null) {
+			return properties.getPrefix() + RabbitCommonProperties.DEAD_LETTER_EXCHANGE;
 		}
 		else {
-			if (partitioned) {
-				String partitionSuffix = "-" + properties.getInstanceIndex();
-				queueName += partitionSuffix;
-			}
-			if (durable) {
-				queue = new Queue(queueName, true, false, false,
-						queueArgs(queueName, properties.getExtension(), false));
-			}
-			else {
-				queue = new Queue(queueName, false, false, true,
-						queueArgs(queueName, properties.getExtension(), false));
-			}
+			return properties.getDeadLetterExchange();
 		}
-		declareQueue(queueName, queue);
-		if (properties.getExtension().isBindQueue()) {
-			declareConsumerBindings(name, properties, exchange, partitioned, queue);
-		}
-		if (durable) {
-			autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName,
-					properties.getExtension());
-		}
-		return queue;
 	}
 
-	private String baseQueueName(String name, String group) {
-		return !StringUtils.hasText(group) ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
-				: groupedName(name, group);
+	@Override
+	protected void afterUnbindConsumer(ConsumerDestination consumerDestination, String group,
+			ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
+		provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName());
 	}
 
 	private MessageRecoverer determineRecoverer(String name, RabbitCommonProperties properties, boolean republish) {
@@ -320,139 +298,6 @@ public class RabbitMessageChannelBinder
 		}
 	}
 
-	@Override
-	protected Exchange createProducerDestinationIfNecessary(String name,
-			ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
-		String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name);
-		Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName);
-		if (producerProperties.getExtension().isDeclareExchange()) {
-			declareExchange(exchangeName, exchange);
-		}
-		return exchange;
-	}
-
-	@Override
-	protected MessageHandler createProducerMessageHandler(final Exchange exchange,
-			ExtendedProducerProperties<RabbitProducerProperties> properties)
-			throws Exception {
-		String prefix = properties.getExtension().getPrefix();
-		String exchangeName = exchange.getName();
-		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()) {
-			if (routingKeyExpression == null) {
-				endpoint.setRoutingKey(destination);
-			}
-			else {
-				endpoint.setRoutingKeyExpressionString(routingKeyExpression);
-			}
-		}
-		else {
-			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, extendedProperties, false));
-				declareQueue(baseQueueName, queue);
-				autoBindDLQ(baseQueueName, baseQueueName, extendedProperties);
-				if (extendedProperties.isBindQueue()) {
-					notPartitionedBinding(exchange, queue, extendedProperties);
-				}
-			}
-			else {
-				// if the stream is partitioned, create one queue for each target partition for the default group
-				for (int i = 0; i < properties.getPartitionCount(); i++) {
-					String partitionSuffix = "-" + i;
-					String partitionQueueName = baseQueueName + partitionSuffix;
-					Queue queue = new Queue(partitionQueueName, true, false, false,
-							queueArgs(partitionQueueName, extendedProperties, false));
-					declareQueue(queue.getName(), queue);
-					autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, extendedProperties);
-					if (extendedProperties.isBindQueue()) {
-						partitionedBinding(destination, exchange, queue, extendedProperties, i);
-					}
-				}
-			}
-		}
-		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper();
-		mapper.setRequestHeaderNames(extendedProperties.getRequestHeaderPatterns());
-		mapper.setReplyHeaderNames(extendedProperties.getReplyHeaderPatterns());
-		endpoint.setHeaderMapper(mapper);
-		endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode());
-		endpoint.setBeanFactory(this.getBeanFactory());
-		endpoint.afterPropertiesSet();
-		return endpoint;
-	}
-
-	private Map<String, Object> queueArgs(String queueName, RabbitCommonProperties properties, boolean isDlq) {
-		Map<String, Object> args = new HashMap<>();
-		if (!isDlq) {
-			if (properties.isAutoBindDlq()) {
-				String dlx;
-				if (properties.getDeadLetterExchange() != null) {
-					dlx = properties.getDeadLetterExchange();
-				}
-				else {
-					dlx = applyPrefix(properties.getPrefix(), "DLX");
-				}
-				args.put("x-dead-letter-exchange", dlx);
-				String dlRk;
-				if (properties.getDeadLetterRoutingKey() != null) {
-					dlRk = properties.getDeadLetterRoutingKey();
-				}
-				else {
-					dlRk = queueName;
-				}
-				args.put("x-dead-letter-routing-key", dlRk);
-			}
-			additionalArgs(args, properties.getExpires(), properties.getMaxLength(), properties.getMaxLengthBytes(),
-					properties.getMaxPriority(), properties.getTtl());
-		}
-		else {
-			if (properties.getDlqDeadLetterExchange() != null) {
-				args.put("x-dead-letter-exchange", properties.getDlqDeadLetterExchange());
-			}
-			if (properties.getDlqDeadLetterRoutingKey() != null) {
-				args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey());
-			}
-			additionalArgs(args, properties.getDlqExpires(), properties.getDlqMaxLength(),
-					properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl());
-		}
-		return args;
-	}
-
-	private void additionalArgs(Map<String, Object> args, Integer expires, Integer maxLength, Integer maxLengthBytes,
-			Integer maxPriority, Integer ttl) {
-		if (expires != null) {
-			args.put("x-expires", expires);
-		}
-		if (maxLength != null) {
-			args.put("x-max-length", maxLength);
-		}
-		if (maxLengthBytes != null) {
-			args.put("x-max-length-bytes", maxLengthBytes);
-		}
-		if (maxPriority != null) {
-			args.put("x-max-priority", maxPriority);
-		}
-		if (ttl != null) {
-			args.put("x-message-ttl", ttl);
-		}
-	}
-
 	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) {
 		RabbitTemplate rabbitTemplate;
 		if (properties.isBatchingEnabled()) {
@@ -476,198 +321,4 @@ public class RabbitMessageChannelBinder
 		return rabbitTemplate;
 	}
 
-	/**
-	 * If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original
-	 * 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 properties the properties.
-	 */
-	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;
-			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);
-			String dlxName = deadLetterExchangeName(properties);
-			final DirectExchange dlx = new DirectExchange(dlxName);
-			declareExchange(dlxName, dlx);
-			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));
-			}
-		}
-	}
-
-	private void declareQueue(String beanName, Queue queue) {
-		try {
-			this.rabbitAdmin.declareQueue(queue);
-		}
-		catch (AmqpConnectException e) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
-			}
-		}
-		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);
-		}
-		catch (AmqpConnectException e) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug(
-						"Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
-			}
-		}
-		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);
-		}
-		catch (AmqpConnectException e) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug(
-						"Declaration of binding: " + rootName + ".binding deferred - connection not available");
-			}
-		}
-		addToAutoDeclareContext(rootName + ".binding", binding);
-	}
-
-	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) {
-		synchronized (this.autoDeclareContext) {
-			if (!this.autoDeclareContext.containsBean(name)) {
-				this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean);
-			}
-		}
-	}
-
-	private void cleanAutoDeclareContext(String prefix, String name) {
-		synchronized (this.autoDeclareContext) {
-			removeSingleton(applyPrefix(prefix, name) + ".binding");
-			removeSingleton(applyPrefix(prefix, name));
-			String dlq = applyPrefix(prefix, name) + ".dlq";
-			removeSingleton(dlq + ".binding");
-			removeSingleton(dlq);
-		}
-	}
-
-	private void removeSingleton(String name) {
-		if (this.autoDeclareContext.containsBean(name)) {
-			ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory();
-			if (beanFactory instanceof DefaultListableBeanFactory) {
-				((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
-			}
-		}
-	}
 }

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

@@ -24,8 +24,10 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
 import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.stream.binder.rabbit.RabbitExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
 import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -61,7 +63,8 @@ public class RabbitMessageChannelBinderConfiguration {
 
 	@Bean
 	RabbitMessageChannelBinder rabbitMessageChannelBinder() {
-		RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory, rabbitProperties);
+		RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory, rabbitProperties,
+				provisioningProvider());
 		binder.setCodec(codec);
 		binder.setAdminAddresses(rabbitBinderConfigurationProperties.getAdminAddresses());
 		binder.setCompressingPostProcessor(gZipPostProcessor());
@@ -82,5 +85,10 @@ public class RabbitMessageChannelBinderConfiguration {
 		gZipPostProcessor.setLevel(rabbitBinderConfigurationProperties.getCompressionLevel());
 		return gZipPostProcessor;
 	}
+
+	@Bean
+	RabbitExchangeQueueProvisioner provisioningProvider() {
+		return new RabbitExchangeQueueProvisioner(rabbitConnectionFactory);
+	}
 }
 

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

@@ -35,6 +35,9 @@ import org.springframework.amqp.rabbit.core.ChannelCallback;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.cloud.stream.binder.AbstractBinder;
+import org.springframework.cloud.stream.binder.rabbit.admin.RabbitAdminException;
+import org.springframework.cloud.stream.binder.rabbit.admin.RabbitBindingCleaner;
+import org.springframework.cloud.stream.binder.rabbit.admin.RabbitManagementUtils;
 import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
 import org.springframework.web.client.RestTemplate;
 import org.springframework.web.util.UriComponentsBuilder;

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

@@ -16,11 +16,6 @@
 
 package org.springframework.cloud.stream.binder.rabbit;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -28,6 +23,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.zip.Deflater;
 
+import com.rabbitmq.http.client.domain.QueueInfo;
 import org.aopalliance.aop.Advice;
 import org.apache.commons.logging.Log;
 import org.junit.Rule;
@@ -59,6 +55,9 @@ import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
 import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
 import org.springframework.cloud.stream.binder.PartitionTestSupport;
 import org.springframework.cloud.stream.binder.Spy;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
+import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
 import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.context.ApplicationContext;
@@ -74,7 +73,10 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
 
-import com.rabbitmq.http.client.domain.QueueInfo;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 /**
  * @author Mark Fisher
@@ -511,7 +513,7 @@ public class RabbitBinderTests extends
 
 		consumerBinding.unbind();
 
-		ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.autoDeclareContext",
+		ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.provisioningProvider.autoDeclareContext",
 				ApplicationContext.class);
 		assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.binding")).isFalse();
 		assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default")).isFalse();
@@ -896,7 +898,9 @@ public class RabbitBinderTests extends
 	public void testLateBinding() throws Exception {
 		RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy();
 		CachingConnectionFactory cf = new CachingConnectionFactory("localhost", proxy.getPort());
-		RabbitMessageChannelBinder rabbitBinder = new RabbitMessageChannelBinder(cf, new RabbitProperties());
+
+		RabbitMessageChannelBinder rabbitBinder = new RabbitMessageChannelBinder(cf, new RabbitProperties(),
+				new RabbitExchangeQueueProvisioner(cf));
 		RabbitTestBinder binder = new RabbitTestBinder(cf, rabbitBinder);
 
 		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();

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

@@ -26,6 +26,10 @@ import org.springframework.cloud.stream.binder.AbstractTestBinder;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
+import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.integration.codec.kryo.PojoCodec;
 import org.springframework.integration.context.IntegrationContextUtils;
@@ -51,7 +55,8 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
 	private final Set<String> exchanges = new HashSet<String>();
 
 	public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
-		this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory, rabbitProperties));
+		this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory, rabbitProperties,
+				new RabbitExchangeQueueProvisioner(connectionFactory)));
 	}
 
 	public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) {