diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java index a7c1b826a..d6297c019 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java +++ b/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; + } + } diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index 2f1b22174..e9c94cb67 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/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. diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index fc4d7a737..be2035a22 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/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()); } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 4ae775194..69c1f21a6 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/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 producerProps = createProducerProperties(); + producerProps.setErrorChannelEnabled(true); + producerProps.getExtension().setConfirmAckChannel("acksChannel"); + Binding producerBinding = binder.bindProducer("acks.0", moduleOutputChannel, producerProps); + final Message message = MessageBuilder.withPayload("acksMessage".getBytes()).build(); + final AtomicReference> 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();