GH-176: Add DLX Provisioning Flexibility

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/176
Resolves #177

- Add `declareDlx` to suppress declaration of DLX
- Add `deadLetterExchangeType` to allow specification of the type

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

@@ -106,6 +106,16 @@ public abstract class RabbitCommonProperties implements MergableProperties {
 	 */
 	private String deadLetterExchange;
 
+	/**
+	 * the type of the DLX, if autoBindDlq is true
+	 */
+	private String deadLetterExchangeType = ExchangeTypes.DIRECT;
+
+	/**
+	 * whether to declare the dead-letter exchange when autoBindDlq is true.
+	 */
+	private boolean declareDlx = true;
+
 	/**
 	 * a dead letter routing key to assign to that queue; if autoBindDlq is true, defaults to destination
 	 */
@@ -297,6 +307,22 @@ public abstract class RabbitCommonProperties implements MergableProperties {
 		this.deadLetterExchange = deadLetterExchange;
 	}
 
+	public String getDeadLetterExchangeType() {
+		return this.deadLetterExchangeType;
+	}
+
+	public void setDeadLetterExchangeType(String deadLetterExchangeType) {
+		this.deadLetterExchangeType = deadLetterExchangeType;
+	}
+
+	public boolean isDeclareDlx() {
+		return this.declareDlx;
+	}
+
+	public void setDeclareDlx(boolean declareDlx) {
+		this.declareDlx = declareDlx;
+	}
+
 	public String getDeadLetterRoutingKey() {
 		return this.deadLetterRoutingKey;
 	}

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

@@ -26,6 +26,7 @@ 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.Binding.DestinationType;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.DirectExchange;
 import org.springframework.amqp.core.Exchange;
@@ -295,16 +296,15 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 			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());
+			if (properties.isDeclareDlx()) {
+				declareExchange(dlxName,
+						new ExchangeBuilder(dlxName, properties.getDeadLetterExchangeType())
+							.durable(true)
+							.build());
 			}
+			Binding dlqBinding = new Binding(dlq.getName(), DestinationType.QUEUE, dlxName,
+					properties.getDlqDeadLetterRoutingKey() == null ? routingKey : properties.getDeadLetterRoutingKey(),
+					null);
 			declareBinding(dlqName, dlqBinding);
 			if (properties instanceof RabbitConsumerProperties &&
 					((RabbitConsumerProperties) properties).isRepublishToDlq()) {
@@ -312,7 +312,8 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 				 *  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));
+				declareBinding(dlqName,
+						new Binding(dlq.getName(), DestinationType.QUEUE, dlxName, baseQueueName, null));
 			}
 		}
 	}

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

@@ -144,11 +144,22 @@ A DLX to assign to the queue.
 Relevant only if `autoBindDlq` is `true`.
 +
 Default: 'prefix+DLX'
+deadLetterExchangeType::
+The type of the DLX to assign to the queue.
+Relevant only if `autoBindDlq` is `true`.
++
+Default: 'direct'
 deadLetterRoutingKey::
 A dead letter routing key to assign to the queue.
 Relevant only if `autoBindDlq` is `true`.
 +
 Default: `destination`
+declareDlx::
+Whether to declare the dead letter exchange for the destination.
+Relevant only if `autoBindDlq` is `true`.
+Set to `false` if you have a pre-configured DLX.
++
+Default: `true`.
 declareExchange::
 Whether to declare the exchange for the destination.
 +
@@ -376,12 +387,25 @@ Relevant only when `autoBindDlq` is `true`.
 Applies only when `requiredGroups` are provided and then only to those groups.
 +
 Default: 'prefix+DLX'
+deadLetterExchangeType::
+The type of the DLX to assign to the queue.
+Relevant only if `autoBindDlq` is `true`.
+Applies only when `requiredGroups` are provided and then only to those groups.
++
+Default: 'direct'
 deadLetterRoutingKey::
 A dead letter routing key to assign to the queue.
 Relevant only when `autoBindDlq` is `true`.
 Applies only when `requiredGroups` are provided and then only to those groups.
 +
 Default: `destination`
+declareDlx::
+Whether to declare the dead letter exchange for the destination.
+Relevant only if `autoBindDlq` is `true`.
+Set to `false` if you have a pre-configured DLX.
+Applies only when `requiredGroups` are provided and then only to those groups.
++
+Default: `true`.
 declareExchange::
 Whether to declare the exchange for the destination.
 +

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

@@ -422,6 +422,7 @@ public class RabbitBinderTests extends
 		extProps.setAutoBindDlq(true);
 		extProps.setDeadLetterQueueName("customDLQ");
 		extProps.setDeadLetterExchange("customDLX");
+		extProps.setDeadLetterExchangeType(ExchangeTypes.TOPIC);
 		extProps.setDeadLetterRoutingKey("customDLRK");
 		extProps.setDlqDeadLetterExchange("propsUser3");
 		extProps.setDlqDeadLetterRoutingKey("propsUser3");
@@ -463,6 +464,16 @@ public class RabbitBinderTests extends
 		assertThat(exchange.isDurable()).isEqualTo(false);
 		assertThat(exchange.isAutoDelete()).isEqualTo(true);
 
+		exchange = rmt.getExchange("customDLX");
+		n = 0;
+		while (n++ < 100 && exchange == null) {
+			Thread.sleep(100);
+			exchange = rmt.getExchange("customDLX");
+		}
+		assertThat(exchange).isInstanceOf(TopicExchange.class);
+		assertThat(exchange.isDurable()).isEqualTo(true);
+		assertThat(exchange.isAutoDelete()).isEqualTo(false);
+
 		QueueInfo queue = rmt.getClient().getQueue("/", "propsUser3.infra");
 		n = 0;
 		while (n++ < 100 && queue == null || queue.getConsumerCount() == 0) {