diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index b81917096a5..bf812fc8bbf 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -48,6 +48,7 @@ * @author Gary Russell * @author Artem Bilan * @author Ngoc Nhan + * @author Jun Cho * * @since 2.1 */ @@ -95,6 +96,12 @@ public void setExpectReply(boolean expectReply) { * received within the confirm timeout or a negative acknowledgment or returned * message is received, an exception will be thrown. Does not apply to the gateway * since it blocks awaiting the reply. + *

+ * When there is no correlation data for the message and the connection factory is + * configured for simple publisher confirms, the message is sent within the scope of a + * {@code RabbitTemplate.invoke()} operation and {@code waitForConfirmsOrDie()} is used + * instead. A returned message does not fail the send in that case; it is published to + * the return channel, if one is configured. * @param waitForConfirm true to block until the confirmation or timeout is received. * @since 5.2 * @see #setConfirmTimeout(long) @@ -156,6 +163,10 @@ protected void endpointInit() { + "does not support returned messages; none will be received"); } } + if (this.waitForConfirm && !usesSimplePublisherConfirms() && !usesCorrelatedPublisherConfirms()) { + this.logger.warn("The 'waitForConfirm' is set to true but the underlying connection factory " + + "does not support publisher confirms; no confirmations will be awaited"); + } Duration confirmTimeout = getConfirmTimeout(); if (confirmTimeout != null) { this.waitForConfirmTimeout = confirmTimeout; @@ -181,6 +192,10 @@ protected void doStop() { multiSend(requestMessage, exchangeName, routingKey); return null; } + else if (this.waitForConfirm && correlationData == null && usesSimplePublisherConfirms()) { + sendAndWaitForConfirms(exchangeName, routingKey, requestMessage); + return null; + } else { send(exchangeName, routingKey, requestMessage, correlationData); if (this.waitForConfirm && correlationData != null) { @@ -208,6 +223,28 @@ private void multiSend(Message requestMessage, @Nullable String exchangeName, }); } + private void sendAndWaitForConfirms(@Nullable String exchangeName, @Nullable String routingKey, + Message requestMessage) { + + RabbitTemplate rabbitTemplateToUse = this.rabbitTemplate; + Assert.notNull(rabbitTemplateToUse, "The 'RabbitTemplate' must be provided to wait for confirms."); + rabbitTemplateToUse.<@Nullable Object>invoke(template -> { + doRabbitSend(exchangeName, routingKey, requestMessage, null, rabbitTemplateToUse); + template.waitForConfirmsOrDie(this.waitForConfirmTimeout.toMillis()); + return null; + }); + } + + private boolean usesSimplePublisherConfirms() { + return this.rabbitTemplate != null + && this.rabbitTemplate.getConnectionFactory().isSimplePublisherConfirms(); + } + + private boolean usesCorrelatedPublisherConfirms() { + return this.rabbitTemplate != null + && this.rabbitTemplate.getConnectionFactory().isPublisherConfirms(); + } + private void waitForConfirm(Message requestMessage, CorrelationData correlationData) { try { Confirm confirm = correlationData.getFuture().get(this.waitForConfirmTimeout.toMillis(), diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd index 14163860e77..7d736888c2d 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd @@ -70,6 +70,11 @@ been received. Requires a template configured for returns. If a confirm is not received within the confirm timeout or a negative acknowledgment or returned message is received, an exception will be thrown. + When there is no correlation data for the message and the connection factory is + configured for simple publisher confirms, the message is sent within the scope of + a RabbitTemplate.invoke() operation and waitForConfirmsOrDie() is used instead. + A returned message does not fail the send in that case; it is published to the + return channel, if one is configured. diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java index 7f74a3a8942..9d3e52f1077 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java @@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test; import org.springframework.amqp.AmqpException; +import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.core.QueueBuilder.Overflow; @@ -53,6 +54,7 @@ /** * @author Gary Russell * @author Artem Bilan + * @author Jun Cho * * @since 5.2 * @@ -63,14 +65,18 @@ public class AmqpOutboundEndpointTests2 implements RabbitTestContainer { static final String QUEUE_TEST_CONFIRM_OK = "testConfirmOk"; + static final String QUEUE_TEST_SIMPLE_CONFIRMS = "testSimpleConfirms"; + @BeforeAll static void initQueue() throws IOException, InterruptedException { RABBITMQ.execInContainer("rabbitmqadmin", "declare", "queue", "name=" + QUEUE_TEST_CONFIRM_OK); + RABBITMQ.execInContainer("rabbitmqadmin", "declare", "queue", "name=" + QUEUE_TEST_SIMPLE_CONFIRMS); } @AfterAll static void deleteQueue() throws IOException, InterruptedException { RABBITMQ.execInContainer("rabbitmqadmin", "delete", "queue", "name=" + QUEUE_TEST_CONFIRM_OK); + RABBITMQ.execInContainer("rabbitmqadmin", "delete", "queue", "name=" + QUEUE_TEST_SIMPLE_CONFIRMS); } @Test @@ -118,6 +124,31 @@ void testWithReject(@Autowired IntegrationFlow flow, @Autowired RabbitAdmin admi admin.deleteQueue(queue.getName()); } + @Test + void simpleConfirmsOk(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired RabbitTemplate template) { + simpleConfirmsFlow.getInputChannel() + .send(new GenericMessage<>("test", Collections.singletonMap("rk", QUEUE_TEST_SIMPLE_CONFIRMS))); + assertThat(template.receive(QUEUE_TEST_SIMPLE_CONFIRMS)).isNotNull(); + } + + @Test + void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow, + @Autowired RabbitAdmin simpleConfirmsAdmin) { + + Queue queue = QueueBuilder.nonDurable().autoDelete().maxLength(1L).overflow(Overflow.rejectPublish).build(); + String queueName = queue.getName(); + simpleConfirmsAdmin.declareQueue(queue); + GenericMessage message = new GenericMessage<>("test", Collections.singletonMap("rk", queueName)); + try { + simpleConfirmsFlow.getInputChannel().send(message); + assertThatThrownBy(() -> simpleConfirmsFlow.getInputChannel().send(message)) + .hasCauseInstanceOf(AmqpIOException.class); + } + finally { + simpleConfirmsAdmin.deleteQueue(queueName); + } + } + @Configuration(proxyBeanMethods = false) @EnableIntegration public static class Config { @@ -138,6 +169,14 @@ public IntegrationFlow flow2(RabbitTemplate template) { .routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class))); } + @Bean + public IntegrationFlow simpleConfirmsFlow(RabbitTemplate simpleConfirmsTemplate) { + return f -> f.handle(Amqp.outboundAdapter(simpleConfirmsTemplate) + .exchangeName("") + .routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class)) + .waitForConfirm(true)); + } + @Bean public CachingConnectionFactory cf() { CachingConnectionFactory ccf = new CachingConnectionFactory(RabbitTestContainer.amqpPort()); @@ -154,6 +193,23 @@ public RabbitTemplate template(ConnectionFactory cf) { return rabbitTemplate; } + @Bean + public CachingConnectionFactory simpleConfirmsCf() { + CachingConnectionFactory ccf = new CachingConnectionFactory(RabbitTestContainer.amqpPort()); + ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE); + return ccf; + } + + @Bean + public RabbitTemplate simpleConfirmsTemplate(ConnectionFactory simpleConfirmsCf) { + return new RabbitTemplate(simpleConfirmsCf); + } + + @Bean + public RabbitAdmin simpleConfirmsAdmin(ConnectionFactory simpleConfirmsCf) { + return new RabbitAdmin(simpleConfirmsCf); + } + @Bean public RabbitAdmin admin(ConnectionFactory cf) { return new RabbitAdmin(cf); diff --git a/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc b/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc index 48fa652acc4..7050ca042f5 100644 --- a/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc +++ b/src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc @@ -123,10 +123,15 @@ Pending confirms are checked every 50% of this value, so the actual time a nack Also see xref:amqp/alternative-confirms-returns.adoc[Alternative Mechanism for Publisher Confirms and Returns]. Default none (nacks will not be generated). <14> When set to true, the calling thread will block, waiting for a publisher confirmation. -This requires a `RabbitTemplate` configured for `confirms` as well as a `confirm-correlation-expression`. +This requires a `RabbitTemplate` configured for `confirms`. +With correlated publisher confirms, the message also needs correlation data, which is usually provided by a `confirm-correlation-expression`. The thread will block for up to `confirm-timeout` (or 5 seconds by default). If a timeout occurs, a `MessageTimeoutException` will be thrown. If returns are enabled and a message is returned, or any other exception occurs while awaiting the confirmation, a `MessageHandlingException` will be thrown, with an appropriate message. +With simple publisher confirms and no correlation data, the adapter sends the message within the scope of a `RabbitTemplate.invoke()` operation and calls `waitForConfirmsOrDie()` instead. +A returned message then does not fail the send; it is published to the `return-channel`, if one is configured. +This option cannot be combined with a `BoundRabbitChannelAdvice` that shares the same `RabbitTemplate`, since both open a scoped operation. +Also see xref:amqp/strict-ordering.adoc[Strict Message Ordering]. <15> The channel to which returned messages are sent. When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter. When there is no `ErrorMessageStrategy` configured, the message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, `amqp_returnRoutingKey`. diff --git a/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc b/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc index 0fc1d3dc821..7b370a2f35a 100644 --- a/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc +++ b/src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc @@ -55,5 +55,9 @@ Notice that the same `RabbitTemplate` (which implements `RabbitOperations`) is u The advice runs the downstream flow within the template's `invoke` method so that all operations run on the same channel. If the optional timeout is provided, when the flow completes, the advice calls the `waitForConfirmsOrDie` method, which throws an exception if the confirmations are not received within the specified time. +NOTE: The outbound adapter opens a scoped operation of its own when `multi-send` is used, and also when `wait-for-confirm` is set with simple publisher confirms. +Do not use those options on an adapter whose flow is already wrapped by a `BoundRabbitChannelAdvice` that shares the same `RabbitTemplate`, because nested scopes are not supported. +The advice already waits for the confirmations of all the messages sent within its scope. + IMPORTANT: There must be no thread hands-off in the downstream flow (`QueueChannel`, `ExecutorChannel`, and others). diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 5a5ef019e10..46d2c5b2b32 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -22,3 +22,11 @@ Java 17 is still the baseline, but Java 25 is supported. The `DefaultErrorMessageStrategy` now preserves the original headers when building the `ErrorMessage`. See xref:error-handling.adoc[] for more information. + +[[x7.2-amqp-changes]] +=== AMQP Support Changes + +The `wait-for-confirm` option on the outbound channel adapter now also works with simple publisher confirms. +When the message has no correlation data, the send is performed within the scope of a `RabbitTemplate.invoke()` operation and `waitForConfirmsOrDie()` is used. +A warning is now logged when the option is set but the connection factory supports no publisher confirms at all. +See xref:amqp/outbound-channel-adapter.adoc[] for more information.