GH-178: Add confirmAckChannel

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/178

Polishing - PR Comments

- use channel name to avoid early instantiation

Resolves #179

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

@@ -76,6 +76,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
 	 */
 	private String routingKeyExpression;
 
+	/**
+	 * the channel name to which to send publisher confirms (acks) if the connection
+	 * factory is so configured; default 'nullChannel'; requires 'errorChannelEnabled=true'
+	 */
+	private String confirmAckChannel;
+
 	/**
 	 * @deprecated - use {@link #setHeaderPatterns(String[])}.
 	 * @param requestHeaderPatterns the patterns.
@@ -177,4 +183,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
 		this.routingKeyExpression = routingKeyExpression;
 	}
 
+	public String getConfirmAckChannel() {
+		return this.confirmAckChannel;
+	}
+
+	public void setConfirmAckChannel(String confirmAckChannel) {
+		this.confirmAckChannel = confirmAckChannel;
+	}
+
 }

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

@@ -376,6 +376,12 @@ compress::
 Whether data should be compressed when sent.
 +
 Default: `false`.
+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.
++
+Default: `nullChannel` (acks are discarded).
 deadLetterQueueName::
 The name of the DLQ
 Only applies if `requiredGroups` are provided and then only to those groups.

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

@@ -66,6 +66,7 @@ import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchang
 import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback;
@@ -76,6 +77,7 @@ import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
 import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.support.DefaultErrorMessageStrategy;
@@ -312,8 +314,15 @@ public class RabbitMessageChannelBinder
 			checkConnectionFactoryIsErrorCapable();
 			endpoint.setReturnChannel(errorChannel);
 			endpoint.setConfirmNackChannel(errorChannel);
-			endpoint.setConfirmAckChannel(getApplicationContext().getBean(
-					IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME, MessageChannel.class));
+			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");
 			endpoint.setErrorMessageStrategy(new DefaultErrorMessageStrategy());
 		}

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

@@ -268,6 +268,31 @@ public class RabbitBinderTests extends
 		producerBinding.unbind();
 	}
 
+	@Test
+	public void testProducerAckChannel() throws Exception {
+		RabbitTestBinder binder = getBinder();
+		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		ccf.setPublisherReturns(true);
+		ccf.setPublisherConfirms(true);
+		ccf.resetConnection();
+		DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties());
+		ExtendedProducerProperties<RabbitProducerProperties> producerProps = createProducerProperties();
+		producerProps.setErrorChannelEnabled(true);
+		producerProps.getExtension().setConfirmAckChannel("acksChannel");
+		Binding<MessageChannel> producerBinding = binder.bindProducer("acks.0", moduleOutputChannel, producerProps);
+		final Message<?> message = MessageBuilder.withPayload("acksMessage".getBytes()).build();
+		final AtomicReference<Message<?>> confirm = new AtomicReference<>();
+		final CountDownLatch confirmLatch = new CountDownLatch(1);
+		binder.getApplicationContext().getBean("acksChannel", DirectChannel.class).subscribe(m -> {
+			confirm.set(m);
+			confirmLatch.countDown();
+		});
+		moduleOutputChannel.send(message);
+		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(confirm.get().getPayload()).isEqualTo("acksMessage".getBytes());
+		producerBinding.unbind();
+	}
+
 	@Test
 	public void testConsumerProperties() throws Exception {
 		RabbitTestBinder binder = getBinder();