GH-303: Add useConfirmHeader

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/303
Resolves #305

docs/src/main/asciidoc/overview.adoc

@@ -667,6 +667,7 @@ confirmAckChannel::
 When `errorChannelEnabled` is true, a channel to which to send positive delivery acknowledgments (aka publisher confirms).
 If the channel does not exist, a `DirectChannel` is registered with this name.
 The connection factory must be configured to enable publisher confirms.
+Mutually exclusive with `useConfirmHeader`.
 +
 Default: `nullChannel` (acks are discarded).
 deadLetterQueueName::
@@ -879,10 +880,131 @@ Default time (in milliseconds) to live to apply to the queue when declared.
 Applies only when `requiredGroups` are provided and then only to those groups.
 +
 Default: `no limit`
+useConfirmHeader::
+See <<publisher-confirms>>.
+Mutually exclusive with `confirmAckChannel`.
++
 
 NOTE: In the case of RabbitMQ, content type headers can be set by external applications.
 Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport -- including transports, such as Kafka (prior to 0.11), that do not natively support headers.
 
+[[publisher-confirms]]
+=== Publisher Confirms
+
+There are two mechanisms to get the result of publishing a message; in each case, the connection factory must have `publisherConfirmType` set `ConfirmType.CORRELATED`.
+The "legacy" mechanism is to set the `confirmAckChannel` to the bean name of a message channel from which you can retrieve the confirmations asynchronously; negative acks are sent to the error channel (if enabled) - see <<rabbit-error-channels>>.
+
+The preferred mechanism, added in version 3.1 is to use a correlation data header and wait for the result via its `Future<Confirm>` property.
+This is particularly useful with a batch listener because you can send multiple messages before waiting for the result.
+To use this technique, set the `useConfirmHeader` property to true
+The following simple application is an example of using this technique:
+
+====
+[source, properties]
+----
+spring.cloud.stream.bindings.input-in-0.group=someGroup
+spring.cloud.stream.bindings.input-in-0.consumer.batch-mode=true
+
+spring.cloud.stream.source=output
+spring.cloud.stream.bindings.output-out-0.producer.error-channel-enabled=true
+
+spring.cloud.stream.rabbit.bindings.output-out-0.producer.useConfirmHeader=true
+spring.cloud.stream.rabbit.bindings.input-in-0.consumer.auto-bind-dlq=true
+spring.cloud.stream.rabbit.bindings.input-in-0.consumer.batch-size=10
+
+spring.rabbitmq.publisher-confirm-type=correlated
+spring.rabbitmq.publisher-returns=true
+----
+====
+
+====
+[source, java]
+----
+@SpringBootApplication
+public class Application {
+
+	private static final Logger log = LoggerFactory.getLogger(Application.class);
+
+	public static void main(String[] args) {
+		SpringApplication.run(Application.class, args);
+	}
+
+	@Autowired
+	private StreamBridge bridge;
+
+	@Bean
+	Consumer<List<String>> input() {
+		return list -> {
+			List<MyCorrelationData> results = new ArrayList<>();
+			list.forEach(str -> {
+				log.info("Received: " + str);
+				MyCorrelationData corr = new MyCorrelationData(UUID.randomUUID().toString(), str);
+				results.add(corr);
+				this.bridge.send("output-out-0", MessageBuilder.withPayload(str.toUpperCase())
+						.setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, corr)
+						.build());
+			});
+			results.forEach(correlation -> {
+				try {
+					Confirm confirm = correlation.getFuture().get(10, TimeUnit.SECONDS);
+					log.info(confirm + " for " + correlation.getPayload());
+					if (correlation.getReturnedMessage() != null) {
+						log.error("Message for " + correlation.getPayload() + " was returned ");
+
+						// try to re-publish, send a DLQ, etc
+
+					}
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					e.printStackTrace();
+				}
+				catch (ExecutionException | TimeoutException e) {
+					e.printStackTrace();
+				}
+			});
+		};
+	}
+
+
+	@Bean
+	public ApplicationRunner runner(BatchingRabbitTemplate template) {
+		return args -> IntStream.range(0, 10).forEach(i ->
+				template.convertAndSend("input-in-0", "input-in-0.rbgh303", "foo" + i));
+	}
+
+	@Bean
+	public BatchingRabbitTemplate template(CachingConnectionFactory cf, TaskScheduler taskScheduler) {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(10, 1000000, 1000);
+		return new BatchingRabbitTemplate(cf, batchingStrategy, taskScheduler);
+	}
+
+}
+
+class MyCorrelationData extends CorrelationData {
+
+	private final String payload;
+
+	MyCorrelationData(String id, String payload) {
+		super(id);
+		this.payload = payload;
+	}
+
+	public String getPayload() {
+		return this.payload;
+	}
+
+}
+----
+====
+
+As you can see, we send each message and then await for the publication results.
+If the messages can't be routed, then correlation data is populated with the returned message before the future is completed.
+
+IMPORTANT: The correlation data must be provided with a unique `id` so that the framework can perform the correlation.
+
+You cannot set both `useConfirmHeader` and `confirmAckChannel` but you can still receive returned messages in the error channel when `useConfirmHeader` is true, but using the correlation header is more convenient.
+
 == Using Existing Queues/Exchanges
 
 By default, the binder will automatically provision a topic exchange with the name being derived from the value of the destination binding property `<prefix><destination>`.
@@ -993,6 +1115,7 @@ RabbitMQ has two types of send failures:
 
 The latter is rare.
 According to the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.".
+You can also get a negative acknowledgment if you publish to a bounded queue with `reject-publish` queue overflow behavior.
 
 As well as enabling producer error channels (as described in "`<<spring-cloud-stream-overview-error-handling>>`"), the RabbitMQ binder only sends messages to the channels if the connection factory is appropriately configured, as follows:
 
@@ -1013,6 +1136,8 @@ The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessa
 * `exchange`: The exchange to which the message was published.
 * `routingKey`: The routing key used when the message was published.
 
+Also see <<publisher-confirms>> for an alternative mechanism to receive returned messages.
+
 For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageException` with the following properties:
 
 * `failedMessage`: The spring-messaging `Message<?>` that failed to be sent.

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

@@ -92,6 +92,15 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
 	 */
 	private String confirmAckChannel;
 
+	/**
+	 * When true, the binding will complete the {@link java.util.concurrent.Future} field
+	 * in a {@link org.springframework.amqp.rabbit.connection.CorrelationData} contained
+	 * in the
+	 * {@link org.springframework.amqp.support.AmqpHeaders#PUBLISH_CONFIRM_CORRELATION}
+	 * header when the confirmation is received.
+	 */
+	private boolean useConfirmHeader;
+
 	/**
 	 * @deprecated - use {@link #setHeaderPatterns(String[])}.
 	 * @param requestHeaderPatterns the patterns.
@@ -209,4 +218,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
 		this.batchingStrategyBeanName = batchingStrategyBeanName;
 	}
 
+	public boolean isUseConfirmHeader() {
+		return this.useConfirmHeader;
+	}
+
+	public void setUseConfirmHeader(boolean useConfirmHeader) {
+		this.useConfirmHeader = useConfirmHeader;
+	}
+
 }

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

@@ -300,11 +300,11 @@ public class RabbitMessageChannelBinder extends
 		String exchangeName = producerDestination.getName();
 		String destination = StringUtils.isEmpty(prefix) ? exchangeName
 				: exchangeName.substring(prefix.length());
-		final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(
-				buildRabbitTemplate(producerProperties.getExtension(),
-						errorChannel != null));
-		endpoint.setExchangeName(producerDestination.getName());
 		RabbitProducerProperties extendedProperties = producerProperties.getExtension();
+		final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(
+				buildRabbitTemplate(extendedProperties,
+						errorChannel != null || extendedProperties.isUseConfirmHeader()));
+		endpoint.setExchangeName(producerDestination.getName());
 		boolean expressionInterceptorNeeded = expressionInterceptorNeeded(
 				extendedProperties);
 		Expression routingKeyExpression = extendedProperties.getRoutingKeyExpression();
@@ -364,19 +364,25 @@ public class RabbitMessageChannelBinder extends
 		if (errorChannel != null) {
 			checkConnectionFactoryIsErrorCapable();
 			endpoint.setReturnChannel(errorChannel);
-			endpoint.setConfirmNackChannel(errorChannel);
-			String ackChannelBeanName = StringUtils
-					.hasText(extendedProperties.getConfirmAckChannel())
-							? extendedProperties.getConfirmAckChannel()
-							: IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME;
-			if (!ackChannelBeanName.equals(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)
-					&& !getApplicationContext().containsBean(ackChannelBeanName)) {
-				GenericApplicationContext context = (GenericApplicationContext) getApplicationContext();
-				context.registerBean(ackChannelBeanName, DirectChannel.class,
-						() -> new DirectChannel());
+			if (!extendedProperties.isUseConfirmHeader()) {
+				endpoint.setConfirmNackChannel(errorChannel);
+				String ackChannelBeanName = StringUtils
+						.hasText(extendedProperties.getConfirmAckChannel())
+								? extendedProperties.getConfirmAckChannel()
+								: IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME;
+				if (!ackChannelBeanName.equals(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)
+						&& !getApplicationContext().containsBean(ackChannelBeanName)) {
+					GenericApplicationContext context = (GenericApplicationContext) getApplicationContext();
+					context.registerBean(ackChannelBeanName, DirectChannel.class,
+							() -> new DirectChannel());
+				}
+				endpoint.setConfirmAckChannelName(ackChannelBeanName);
+				endpoint.setConfirmCorrelationExpressionString("#root");
+			}
+			else {
+				Assert.state(!StringUtils.hasText(extendedProperties.getConfirmAckChannel()),
+						"You cannot specify a 'confirmAckChannel' when 'useConfirmHeader' is true");
 			}
-			endpoint.setConfirmAckChannelName(ackChannelBeanName);
-			endpoint.setConfirmCorrelationExpressionString("#root");
 			endpoint.setErrorMessageStrategy(new DefaultErrorMessageStrategy());
 		}
 		endpoint.setHeadersMappedLast(true);
@@ -861,8 +867,7 @@ public class RabbitMessageChannelBinder extends
 				consumerProperties);
 	}
 
-	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties,
-			boolean mandatory) {
+	private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) {
 		RabbitTemplate rabbitTemplate;
 		if (properties.isBatchingEnabled()) {
 			BatchingStrategy batchingStrategy = getBatchingStrategy(properties);

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

@@ -64,6 +64,8 @@ import org.springframework.amqp.rabbit.batch.MessageBatch;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
 import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -245,7 +247,7 @@ public class RabbitBinderTests extends
 		RabbitTestBinder binder = getBinder();
 		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
 		ccf.setPublisherReturns(true);
-		ccf.setPublisherConfirms(true);
+		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
 		ccf.resetConnection();
 		DirectChannel moduleOutputChannel = createBindableChannel("output",
 				new BindingProperties());
@@ -326,7 +328,7 @@ public class RabbitBinderTests extends
 		RabbitTestBinder binder = getBinder();
 		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
 		ccf.setPublisherReturns(true);
-		ccf.setPublisherConfirms(true);
+		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
 		ccf.resetConnection();
 		DirectChannel moduleOutputChannel = createBindableChannel("output",
 				new BindingProperties());
@@ -350,6 +352,30 @@ public class RabbitBinderTests extends
 		producerBinding.unbind();
 	}
 
+	@Test
+	public void testProducerConfirmHeader() throws Exception {
+		RabbitTestBinder binder = getBinder();
+		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		ccf.setPublisherReturns(true);
+		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
+		ccf.resetConnection();
+		DirectChannel moduleOutputChannel = createBindableChannel("output",
+				new BindingProperties());
+		ExtendedProducerProperties<RabbitProducerProperties> producerProps = createProducerProperties();
+		producerProps.getExtension().setUseConfirmHeader(true);
+		Binding<MessageChannel> producerBinding = binder.bindProducer("confirms.0",
+				moduleOutputChannel, producerProps);
+		CorrelationData correlation = new CorrelationData("testConfirm");
+		final Message<?> message = MessageBuilder.withPayload("confirmsMessage".getBytes())
+				.setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, correlation)
+				.build();
+		moduleOutputChannel.send(message);
+		Confirm confirm = correlation.getFuture().get(10, TimeUnit.SECONDS);
+		assertThat(confirm.isAck()).isTrue();
+		assertThat(correlation.getReturnedMessage()).isNotNull();
+		producerBinding.unbind();
+	}
+
 	@Test
 	public void testConsumerProperties() throws Exception {
 		RabbitTestBinder binder = getBinder();