diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java index cfb5ed0a7d..e5fce82f0d 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java @@ -81,6 +81,7 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "wait-for-confirm"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression", diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java index 9f24afa780..2de73b5dbe 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpOutboundEndpointSpec.java @@ -45,4 +45,16 @@ public class AmqpOutboundEndpointSpec return super.mappedReplyHeaders(headers); } + /** + * Wait for a publisher confirm. + * @param waitForConfirm true to wait. + * @return the spec. + * @since 5.2 + * @see AmqpOutboundEndpoint#setWaitForConfirm(boolean) + */ + public AmqpOutboundEndpointSpec waitForConfirm(boolean waitForConfirm) { + this.target.setWaitForConfirm(waitForConfirm); + return this; + } + } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java index ddf82354f1..582669d0be 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -402,6 +402,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin return this.headersMappedLast; } + @Nullable protected Duration getConfirmTimeout() { return this.confirmTimeout; } 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 be23be182a..b51ea63345 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 @@ -16,12 +16,21 @@ package org.springframework.integration.amqp.outbound; +import java.time.Duration; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.context.Lifecycle; +import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.amqp.support.MappingUtils; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; @@ -38,7 +47,9 @@ import org.springframework.util.Assert; * @since 2.1 */ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint - implements RabbitTemplate.ConfirmCallback, ReturnCallback { + implements ConfirmCallback, ReturnCallback { + + private static final Duration DEFAULT_CONFIRM_TIMEOUT = Duration.ofSeconds(5); private final AmqpTemplate amqpTemplate; @@ -46,6 +57,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint private boolean expectReply; + private boolean waitForConfirm; + + private Duration waitForConfirmTimeout = DEFAULT_CONFIRM_TIMEOUT; + public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) { Assert.notNull(amqpTemplate, "amqpTemplate must not be null"); this.amqpTemplate = amqpTemplate; @@ -62,6 +77,19 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint this.expectReply = expectReply; } + /** + * Set to true if you want to block the calling thread until a publisher confirm has + * 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. Does not apply to the gateway + * since it blocks awaiting the reply. + * @param waitForConfirm true to block until the confirmation or timeout is received. + * @since 5.2 + * @see #setConfirmTimeout(long) + */ + public void setWaitForConfirm(boolean waitForConfirm) { + this.waitForConfirm = waitForConfirm; + } @Override public String getComponentType() { @@ -86,6 +114,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint "RabbitTemplate implementation is required for publisher confirms"); this.rabbitTemplate.setReturnCallback(this); } + Duration confirmTimeout = getConfirmTimeout(); + if (confirmTimeout != null) { + this.waitForConfirmTimeout = confirmTimeout; + } } @Override @@ -101,14 +133,39 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint String exchangeName = generateExchangeName(requestMessage); String routingKey = generateRoutingKey(requestMessage); if (this.expectReply) { - return this.sendAndReceive(exchangeName, routingKey, requestMessage, correlationData); + return sendAndReceive(exchangeName, routingKey, requestMessage, correlationData); } else { - this.send(exchangeName, routingKey, requestMessage, correlationData); + send(exchangeName, routingKey, requestMessage, correlationData); + if (this.waitForConfirm && correlationData != null) { + waitForConfirm(requestMessage, correlationData); + } return null; } } + private void waitForConfirm(Message requestMessage, CorrelationData correlationData) { + try { + Confirm confirm = correlationData.getFuture().get(this.waitForConfirmTimeout.toMillis(), + TimeUnit.MILLISECONDS); + if (!confirm.isAck()) { + throw new AmqpException("Negative publisher confirm received: " + confirm); + } + if (correlationData.getReturnedMessage() != null) { + throw new AmqpException("Message was returned by the broker"); + } + } + catch (@SuppressWarnings("unused") InterruptedException e) { + Thread.currentThread().interrupt(); + } + catch (ExecutionException e) { + throw new AmqpException("Failed to get publisher confirm", e); + } + catch (TimeoutException e) { + throw new MessageTimeoutException(requestMessage, this + ": Timed out awaiting publisher confirm", e); + } + } + private void send(String exchangeName, String routingKey, final Message requestMessage, CorrelationData correlationData) { if (this.rabbitTemplate != null) { diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.2.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.2.xsd index a2a311f24d..10614e6536 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.2.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.2.xsd @@ -58,6 +58,21 @@ + + + + + Set to true if you want to block the calling thread until a publisher confirm has + 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. + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml index cabb3c3405..24a2199442 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml @@ -65,6 +65,7 @@ confirm-ack-channel="ackChannel" confirm-nack-channel="nackChannel" confirm-timeout="2000" + wait-for-confirm="true" error-message-strategy="ems"/> diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java index c99f4d18a3..205ac93466 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java @@ -178,6 +178,7 @@ public class AmqpOutboundChannelAdapterParserTests { assertThat(TestUtils.getPropertyValue(endpoint, "confirmAckChannel")).isSameAs(ackChannel); assertThat(TestUtils.getPropertyValue(endpoint, "confirmNackChannel")).isSameAs(nullChannel); assertThat(TestUtils.getPropertyValue(endpoint, "errorMessageStrategy")).isSameAs(context.getBean("ems")); + assertThat(TestUtils.getPropertyValue(endpoint, "waitForConfirm", Boolean.class)).isFalse(); } @Test @@ -191,6 +192,7 @@ public class AmqpOutboundChannelAdapterParserTests { assertThat(TestUtils.getPropertyValue(endpoint, "confirmNackChannel")).isSameAs(nackChannel); assertThat(TestUtils.getPropertyValue(endpoint, "confirmTimeout")).isEqualTo(Duration.ofMillis(2000)); assertThat(TestUtils.getPropertyValue(endpoint, "errorMessageStrategy")).isSameAs(context.getBean("ems")); + assertThat(TestUtils.getPropertyValue(endpoint, "waitForConfirm", Boolean.class)).isTrue(); } @SuppressWarnings("rawtypes") 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 new file mode 100644 index 0000000000..b70b0de08f --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java @@ -0,0 +1,130 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.outbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.Collections; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.QueueBuilder; +import org.springframework.amqp.core.QueueBuilder.Overflow; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.RabbitAvailable; +import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.amqp.dsl.Amqp; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.DisabledIf; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 5.2 + * + */ +@SpringJUnitConfig +@RabbitAvailable(queues = "testConfirmOk") +@DirtiesContext +public class AmqpOutboundEndpointTests2 { + + @Test + void testConfirmOk(@Autowired IntegrationFlow flow, @Autowired RabbitTemplate template) { + flow.getInputChannel().send(new GenericMessage<>("test", Collections.singletonMap("rk", "testConfirmOk"))); + assertThat(template.receive("testConfirmOk")).isNotNull(); + } + + @Test + void testWithReturn(@Autowired IntegrationFlow flow) { + assertThatThrownBy(() -> flow.getInputChannel() + .send(new GenericMessage<>("test", Collections.singletonMap("rk", "junkjunk")))) + .isInstanceOf(MessageHandlingException.class) + .hasCauseInstanceOf(AmqpException.class) + .extracting(ex -> ex.getCause()) + .extracting(ex -> ex.getMessage()) + .isEqualTo("Message was returned by the broker"); + } + + @Test + @DisabledIf("#{systemEnvironment['TRAVIS'] ?: false}") // needs RabbitMQ 3.7 + void testWithReject(@Autowired IntegrationFlow flow, @Autowired RabbitAdmin admin, + @Autowired RabbitTemplate template) { + + Queue queue = QueueBuilder.nonDurable().autoDelete().maxLength(1).overflow(Overflow.rejectPublish).build(); + admin.declareQueue(queue); + flow.getInputChannel().send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName()))); + assertThatThrownBy(() -> flow.getInputChannel() + .send(new GenericMessage<>("test", Collections.singletonMap("rk", queue.getName())))) + .hasCauseInstanceOf(AmqpException.class) + .extracting(ex -> ex.getCause()) + .extracting(ex -> ex.getMessage()) + .matches(msg -> msg.matches("Negative publisher confirm received: .*")); + assertThat(template.receive(queue.getName())).isNotNull(); + admin.deleteQueue(queue.getName()); + } + + @Configuration(proxyBeanMethods = false) + @EnableIntegration + public static class Config { + + @Bean + public IntegrationFlow flow(RabbitTemplate template) { + return f -> f.handle(Amqp.outboundAdapter(template) + .exchangeName("") + .routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class)) + .confirmCorrelationFunction(msg -> msg) + .waitForConfirm(true)); + } + + @Bean + public CachingConnectionFactory cf() { + CachingConnectionFactory ccf = new CachingConnectionFactory( + RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); + ccf.setPublisherConfirms(true); + ccf.setPublisherReturns(true); + return ccf; + } + + @Bean + public RabbitTemplate template(ConnectionFactory cf) { + RabbitTemplate rabbitTemplate = new RabbitTemplate(cf); + rabbitTemplate.setMandatory(true); + rabbitTemplate.setReceiveTimeout(10_000); + return rabbitTemplate; + } + + @Bean + public RabbitAdmin admin(ConnectionFactory cf) { + return new RabbitAdmin(cf); + } + + } + +} diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index e0736fb16a..b50808da43 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -547,12 +547,12 @@ The following example shows the available properties for an AMQP outbound channe confirm-ack-channel="" <11> confirm-nack-channel="" <12> confirm-timeout="" <13> - return-channel="" <14> - error-message-strategy="" <15> - header-mapper="" <16> - mapped-request-headers="" <17> - lazy-connect="true" /> <18> - + wait-for-confirm="" <14> + return-channel="" <15> + error-message-strategy="" <16> + header-mapper="" <17> + mapped-request-headers="" <18> + lazy-connect="true" /> <19> ---- <1> The unique ID for this adapter. @@ -609,21 +609,26 @@ Optional (the default is `nullChannel`). <13> When set, the adapter will synthesize a negative acknowledgment (nack) if a publisher confirm is not received within this time in milliseconds. Pending confirms are checked every 50% of this value, so the actual time a nack is sent will be between 1x and 1.5x this value. Default none (nacks will not be generated). -<14> The channel to which returned messages are sent. +<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`. +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 confirm, a `MessageHandlingException` will be thrown, with an appropriate message. +<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`. When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `ReturnedAmqpMessageException` payload. Optional. -<15> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages. -<16> A reference to an `AmqpHeaderMapper` to use when sending AMQP Messages. +<16> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages. +<17> A reference to an `AmqpHeaderMapper` to use when sending AMQP Messages. By default, only standard AMQP properties (such as `contentType`) are copied to the Spring Integration `MessageHeaders`. Any user-defined headers is not copied to the message by the default`DefaultAmqpHeaderMapper`. Not allowed if 'request-header-names' is provided. Optional. -<17> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message. +<18> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message. Not allowed if the 'header-mapper' reference is provided. The values in this list can also be simple patterns to be matched against the header names (e.g. `"\*"` or `"thing1*, thing2"` or `"*thing1"`). -<18> When set to `false`, the endpoint attempts to connect to the broker during application context initialization. +<19> When set to `false`, the endpoint attempts to connect to the broker during application context initialization. This allows "`fail fast`" detection of bad configuration but also causes initialization to fail if the broker is down. When `true` (the default), the connection is established (if it does not already exist because some other component established it) when the first message is sent. ==== diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index ac25309819..3ea3339183 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -89,6 +89,9 @@ See <<./amqp.adoc#amqp-outbound-endpoints,Outbound Channel Adapter>> for more in The inbound channel adapter can now receive batched messages as a `List` payload instead of receiving a discrete message for each batch fragment. See <<./amqp.adoc#amqp-debatching,Batched Messages>> for more information. +The outbound channel adapter can now be configured to block the calling thread until a publisher confirm (acknowledgment) is received. +See <<./amqp.adoc#amqp-outbound-channel-adapter,Outbound Channel Adapter>> for more information. + [[x5.2-file]] ==== File Changes