GH-53: Add 'republishDeliveryMode'

Resolves #53

Add a property to configure the delivery mode of republished DLQ messages.

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.rabbit.properties;
 import javax.validation.constraints.Min;
 
 import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.util.Assert;
 
 /**
@@ -41,6 +42,8 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 
 	private boolean republishToDlq;
 
+	private MessageDeliveryMode republishDeliveyMode = MessageDeliveryMode.PERSISTENT;
+
 	private boolean requeueRejected = false;
 
 	private String[] headerPatterns = new String[] {"*"};
@@ -60,7 +63,7 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	}
 
 	public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
-		Assert.notNull("Acknowledge mode cannot be null");
+		Assert.notNull(acknowledgeMode, "Acknowledge mode cannot be null");
 		this.acknowledgeMode = acknowledgeMode;
 	}
 
@@ -129,6 +132,14 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 		return requeueRejected;
 	}
 
+	public MessageDeliveryMode getRepublishDeliveyMode() {
+		return this.republishDeliveyMode;
+	}
+
+	public void setRepublishDeliveyMode(MessageDeliveryMode republishDeliveyMode) {
+		this.republishDeliveyMode = republishDeliveyMode;
+	}
+
 	public void setRequeueRejected(boolean requeueRejected) {
 		this.requeueRejected = requeueRejected;
 	}

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

@@ -44,6 +44,7 @@ If retry is enabled (`maxAttempts > 1`) failed messages will be delivered to the
 If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to `false` (default) so that a failed message will be routed to the DLQ, instead of being requeued.
 In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it); this enables additional information to be added to the message in headers, such as the stack trace in the `x-exception-stacktrace` header.
 This option does not need retry enabled; you can republish a failed message after just one attempt.
+Starting with _version 1.2_, you can configure the delivery mode of republished messsages; see property `republishDeliveryMode`.
 
 IMPORTANT: Setting `requeueRejected` to `true` will cause the message to be requeued and redelivered continually, which is likely not what you want unless the failure issue is transient.
 In general, it's better to enable retry within the binder by setting `maxAttempts` to greater than one, or set `republishToDlq` to `true`.
@@ -220,6 +221,10 @@ requeueRejected::
   Whether delivery failures should be requeued when retry is disabled or republishToDlq is false.
 +
 Default: `false`.
+republishDeliveryMode::
+  When `republishToDlq` is `true`, specify the delivery mode of the republished message.
++
+Default: `DeliveryMode.PERSISTENT`
 republishToDlq::
   By default, messages which fail after retries are exhausted are rejected.
 If a dead-letter queue (DLQ) is configured, RabbitMQ will route the failed message (unchanged) to the DLQ.

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

@@ -20,6 +20,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
+import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
@@ -259,8 +260,7 @@ public class RabbitMessageChannelBinder
 		if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) {
 			RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
 					.retryOperations(buildRetryTemplate(properties))
-					.recoverer(determineRecoverer(baseQueueName, properties.getExtension(),
-							properties.getExtension().isRepublishToDlq()))
+					.recoverer(determineRecoverer(baseQueueName, properties.getExtension()))
 					.build();
 			listenerContainer.setAdviceChain(retryInterceptor);
 		}
@@ -293,12 +293,27 @@ public class RabbitMessageChannelBinder
 		provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName());
 	}
 
-	private MessageRecoverer determineRecoverer(String name, RabbitCommonProperties properties, boolean republish) {
-		if (republish) {
+	private MessageRecoverer determineRecoverer(String name, final RabbitConsumerProperties properties) {
+		if (properties.isRepublishToDlq()) {
 			RabbitTemplate errorTemplate = new RabbitTemplate(this.connectionFactory);
-			return new RepublishMessageRecoverer(errorTemplate,
-					deadLetterExchangeName(properties),
-					applyPrefix(properties.getPrefix(), name));
+			if (properties.getRepublishDeliveyMode() != null) {
+				return new RepublishMessageRecoverer(errorTemplate,
+						deadLetterExchangeName(properties),
+						applyPrefix(properties.getPrefix(), name)) {
+
+							@Override
+							public void recover(Message message, Throwable cause) {
+								message.getMessageProperties().setDeliveryMode(properties.getRepublishDeliveyMode());
+								super.recover(message, cause);
+							}
+
+				};
+			}
+			else {
+				return new RepublishMessageRecoverer(errorTemplate,
+						deadLetterExchangeName(properties),
+						applyPrefix(properties.getPrefix(), name));
+			}
 		}
 		else {
 			return new RejectAndDontRequeueRecoverer();

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

@@ -636,6 +636,7 @@ public class RabbitBinderTests extends
 		properties.getExtension().setPrefix("bindertest.");
 		properties.getExtension().setAutoBindDlq(true);
 		properties.getExtension().setRepublishToDlq(true);
+		properties.getExtension().setRepublishDeliveyMode(MessageDeliveryMode.NON_PERSISTENT);
 		properties.setMaxAttempts(1); // disable retry
 		properties.setPartitioned(true);
 		properties.setInstanceIndex(0);
@@ -706,6 +707,8 @@ public class RabbitBinderTests extends
 		assertThat(received.getMessageProperties().getHeaders().get("x-original-routingKey"))
 				.isEqualTo("partPubDLQ.0-1");
 		assertThat(received.getMessageProperties().getHeaders()).doesNotContainKey(BinderHeaders.PARTITION_HEADER);
+		assertThat(received.getMessageProperties().getReceivedDeliveryMode())
+				.isEqualTo(MessageDeliveryMode.NON_PERSISTENT);
 
 		output.send(new GenericMessage<>(0));
 		received = template.receive(streamDLQName);
@@ -800,6 +803,7 @@ public class RabbitBinderTests extends
 		assertThat(received.getMessageProperties().getReceivedRoutingKey())
 				.isEqualTo("bindertest.partDLQ.1.dlqPartGrp-1");
 		assertThat(received.getMessageProperties().getHeaders()).doesNotContainKey(BinderHeaders.PARTITION_HEADER);
+		assertThat(received.getMessageProperties().getReceivedDeliveryMode()).isEqualTo(MessageDeliveryMode.PERSISTENT);
 
 		output.send(new GenericMessage<Integer>(0));
 		received = template.receive(streamDLQName);