From 5245effd3247018a7246ea06231390638e4837f9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 11 Mar 2019 15:04:57 -0400 Subject: [PATCH] GH-2797: AMQP: Add confirm-timeout Resolves https://github.com/spring-projects/spring-integration/issues/2797 Avoid runtime casting to `RabbitTemplate`. --- .../AmqpOutboundChannelAdapterParser.java | 3 +- .../config/AmqpOutboundGatewayParser.java | 3 +- .../AbstractAmqpOutboundEndpoint.java | 53 +++++++++++++-- .../amqp/outbound/AmqpOutboundEndpoint.java | 42 ++++++++---- .../outbound/AsyncAmqpOutboundGateway.java | 8 ++- .../config/spring-integration-amqp-5.2.xsd | 8 +++ ...boundChannelAdapterParserTests-context.xml | 17 +++++ ...AmqpOutboundChannelAdapterParserTests.java | 22 +++++-- .../AmqpOutboundGatewayParserTests.java | 16 ++--- .../outbound/AmqpOutboundEndpointTests.java | 56 +++++++++++++++- .../amqp/outbound/AsyncAmqpGatewayTests.java | 2 +- src/reference/asciidoc/amqp.adoc | 65 ++++++++++++------- src/reference/asciidoc/whats-new.adoc | 6 ++ 13 files changed, 243 insertions(+), 58 deletions(-) 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 30aee77cb4..0c5091885e 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-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. @@ -80,6 +80,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.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/config/AmqpOutboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java index 6ae5b81531..a7579ddcb8 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-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. @@ -94,6 +94,7 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-timeout"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); BeanDefinitionBuilder mapperBuilder = BeanDefinitionBuilder 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 5ba1806b3c..4ee398b71c 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 @@ -16,14 +16,18 @@ package org.springframework.integration.amqp.outbound; +import java.time.Duration; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ScheduledFuture; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.beans.factory.BeanFactory; @@ -40,6 +44,7 @@ import org.springframework.integration.handler.ExpressionEvaluatingMessageProces import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -100,8 +105,12 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + private Duration confirmTimeout; + private volatile boolean running; + private volatile ScheduledFuture confirmChecker; + /** * Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers. * Defaults to {@link DefaultAmqpHeaderMapper#outboundMapper()}. @@ -311,6 +320,18 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin this.errorMessageStrategy = errorMessageStrategy; } + /** + * Set a timeout after which a nack will be synthesized if no publisher confirm has + * been received within that time. Missing confirms will be checked every 50% of this + * value so the synthesized nack will be sent between 1x and 1.5x this timeout. + * @param confirmTimeout the approximate timeout. + * @since 5.2 + * @see #setConfirmNackChannel(MessageChannel) + */ + public void setConfirmTimeout(long confirmTimeout) { + this.confirmTimeout = Duration.ofMillis(confirmTimeout); + } + protected final synchronized void setConnectionFactory(ConnectionFactory connectionFactory) { this.connectionFactory = connectionFactory; } @@ -381,6 +402,10 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin return this.headersMappedLast; } + protected Duration getConfirmTimeout() { + return this.confirmTimeout; + } + @Override protected final void doInit() { Assert.state(this.exchangeNameExpression == null || this.exchangeName == null, @@ -411,7 +436,8 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin } else { NullChannel nullChannel = extractTypeIfPossible(this.confirmAckChannel, NullChannel.class); - Assert.state((this.confirmAckChannel == null || nullChannel != null) && this.confirmAckChannelName == null, + Assert.state( + (this.confirmAckChannel == null || nullChannel != null) && this.confirmAckChannelName == null, "A 'confirmCorrelationExpression' is required when specifying a 'confirmAckChannel'"); nullChannel = extractTypeIfPossible(this.confirmNackChannel, NullChannel.class); Assert.state( @@ -450,16 +476,35 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin } } doStart(); + if (this.confirmTimeout != null && getConfirmNackChannel() != null && getRabbitTemplate() != null) { + this.confirmChecker = getTaskScheduler() + .scheduleAtFixedRate(checkUnconfirmed(), this.confirmTimeout.dividedBy(2L)); + } this.running = true; } } + private Runnable checkUnconfirmed() { + return () -> { + Collection unconfirmed = + getRabbitTemplate().getUnconfirmed(getConfirmTimeout().toMillis()); + unconfirmed.forEach(correlation -> handleConfirm(correlation, false, "Confirm timed out")); + }; + } + + @Nullable + protected abstract RabbitTemplate getRabbitTemplate(); + @Override public synchronized void stop() { if (this.running) { doStop(); } this.running = false; + if (this.confirmChecker != null) { + this.confirmChecker.cancel(false); + this.confirmChecker = null; + } } protected void doStart() { @@ -526,7 +571,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin } protected Message buildReturnedMessage(org.springframework.amqp.core.Message message, - int replyCode, String replyText, String exchange, String routingKey, MessageConverter converter) { + int replyCode, String replyText, String exchange, String returnedRoutingKey, MessageConverter converter) { Object returnedObject = converter.fromMessage(message); AbstractIntegrationMessageBuilder builder = (returnedObject instanceof Message) ? this.getMessageBuilderFactory().fromMessage((Message) returnedObject) @@ -537,12 +582,12 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin .setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode) .setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText) .setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange) - .setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey); + .setHeader(AmqpHeaders.RETURN_ROUTING_KEY, returnedRoutingKey); } Message returnedMessage = builder.build(); if (this.errorMessageStrategy != null) { returnedMessage = this.errorMessageStrategy.buildErrorMessage(new ReturnedAmqpMessageException( - returnedMessage, message, replyCode, replyText, exchange, routingKey), null); + returnedMessage, message, replyCode, replyText, exchange, returnedRoutingKey), null); } return returnedMessage; } 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 3b724739b9..6e72c62f79 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-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. @@ -42,13 +42,19 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint private final AmqpTemplate amqpTemplate; - private volatile boolean expectReply; + private final RabbitTemplate rabbitTemplate; + + private boolean expectReply; public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) { Assert.notNull(amqpTemplate, "amqpTemplate must not be null"); this.amqpTemplate = amqpTemplate; if (amqpTemplate instanceof RabbitTemplate) { setConnectionFactory(((RabbitTemplate) amqpTemplate).getConnectionFactory()); + this.rabbitTemplate = (RabbitTemplate) amqpTemplate; + } + else { + this.rabbitTemplate = null; } } @@ -62,17 +68,23 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint return this.expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter"; } + + @Override + protected RabbitTemplate getRabbitTemplate() { + return this.rabbitTemplate; + } + @Override protected void endpointInit() { if (getConfirmCorrelationExpression() != null) { - Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate, + Assert.notNull(this.rabbitTemplate, "RabbitTemplate implementation is required for publisher confirms"); - ((RabbitTemplate) this.amqpTemplate).setConfirmCallback(this); + this.rabbitTemplate.setConfirmCallback(this); } if (getReturnChannel() != null) { - Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate, + Assert.notNull(this.rabbitTemplate, "RabbitTemplate implementation is required for publisher confirms"); - ((RabbitTemplate) this.amqpTemplate).setReturnCallback(this); + this.rabbitTemplate.setReturnCallback(this); } } @@ -99,12 +111,12 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint private void send(String exchangeName, String routingKey, final Message requestMessage, CorrelationData correlationData) { - if (this.amqpTemplate instanceof RabbitTemplate) { - MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); + if (this.rabbitTemplate != null) { + MessageConverter converter = this.rabbitTemplate.getMessageConverter(); org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast()); addDelayProperty(requestMessage, amqpMessage); - ((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData); + this.rabbitTemplate.send(exchangeName, routingKey, amqpMessage, correlationData); } else { this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(), @@ -118,14 +130,15 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint private AbstractIntegrationMessageBuilder sendAndReceive(String exchangeName, String routingKey, Message requestMessage, CorrelationData correlationData) { - Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate, + + Assert.state(this.rabbitTemplate != null, "RabbitTemplate implementation is required for publisher confirms"); - MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); + MessageConverter converter = this.rabbitTemplate.getMessageConverter(); org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast()); addDelayProperty(requestMessage, amqpMessage); org.springframework.amqp.core.Message amqpReplyMessage = - ((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage, + this.rabbitTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage, correlationData); if (amqpReplyMessage == null) { @@ -142,8 +155,9 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint @Override public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) { - // safe to cast; we asserted we have a RabbitTemplate in doInit() - MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); + + // no need for null check; we asserted we have a RabbitTemplate in doInit() + MessageConverter converter = this.rabbitTemplate.getMessageConverter(); Message returned = buildReturnedMessage(message, replyCode, replyText, exchange, routingKey, converter); getReturnChannel().send(returned); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index 3f9af15002..a831eb5a61 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-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. @@ -21,6 +21,7 @@ import org.springframework.amqp.core.AmqpReplyTimeoutException; import org.springframework.amqp.rabbit.AsyncRabbitTemplate; import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.integration.amqp.support.MappingUtils; import org.springframework.integration.handler.ReplyRequiredException; @@ -61,6 +62,11 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { return "amqp:outbound-async-gateway"; } + @Override + protected RabbitTemplate getRabbitTemplate() { + return this.template.getRabbitTemplate(); + } + @Override protected void doStart() { super.doStart(); 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 877887067b..08d6662a67 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 @@ -544,6 +544,14 @@ property set to TRUE. + + + + + + + + + @@ -62,6 +75,10 @@ + + + + { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index 8dae4a0816..9d39de31c2 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -109,9 +109,9 @@ public class AmqpOutboundGatewayParserTests { assertThat(TestUtils.getPropertyValue(endpoint, "requiresReply", Boolean.class)).isFalse(); assertThat(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)).isTrue(); - Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate"); + Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate"); amqpTemplateField.setAccessible(true); - RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); + RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); final AtomicBoolean shouldBePersistent = new AtomicBoolean(); @@ -172,9 +172,9 @@ public class AmqpOutboundGatewayParserTests { assertThat(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode")).isNull(); assertThat(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)).isFalse(); - Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate"); + Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate"); amqpTemplateField.setAccessible(true); - RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); + RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); Mockito.doAnswer(invocation -> { @@ -221,9 +221,9 @@ public class AmqpOutboundGatewayParserTests { AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class); - Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate"); + Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate"); amqpTemplateField.setAccessible(true); - RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); + RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); Mockito.doAnswer(invocation -> { @@ -270,9 +270,9 @@ public class AmqpOutboundGatewayParserTests { AmqpOutboundEndpoint endpoint = (AmqpOutboundEndpoint) chainHandlers.get(0); - Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate"); + Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "rabbitTemplate"); amqpTemplateField.setAccessible(true); - RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); + RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "rabbitTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); Mockito.doAnswer(invocation -> { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java index 66833b74e3..f83161486b 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java @@ -17,12 +17,22 @@ package org.springframework.integration.amqp.outbound; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.BDDMockito.willDoNothing; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.TimeUnit; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -31,15 +41,21 @@ import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.amqp.utils.test.TestUtils; +import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.amqp.support.NackedAmqpMessageException; import org.springframework.integration.amqp.support.ReturnedAmqpMessageException; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.mapping.support.JsonHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.DirtiesContext.ClassMode; import org.springframework.test.context.ContextConfiguration; @@ -133,7 +149,7 @@ public class AmqpOutboundEndpointTests { } @Test - public void adapterWithPublisherConfirms() throws Exception { + public void adapterWithPublisherConfirms() { Message message = MessageBuilder.withPayload("hello") .setHeader("amqp_confirmCorrelationData", "foo") .build(); @@ -144,6 +160,42 @@ public class AmqpOutboundEndpointTests { assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE); } + @Test + public void syncConfirmTimeout() { + Message message = new GenericMessage<>("foo"); + RabbitTemplate template = spy(RabbitTemplate.class); + willDoNothing().given(template).send(isNull(), isNull(), any(), any()); + List correlationList = new ArrayList<>(); + willReturn(correlationList).given(template).getUnconfirmed(100L); + ArgumentCaptor correlationCaptor = ArgumentCaptor.forClass(CorrelationData.class); + AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(template); + PollableChannel nacks = new QueueChannel(); + endpoint.setConfirmNackChannel(nacks); + endpoint.setConfirmCorrelationExpressionString("headers.id"); + endpoint.setBeanFactory(mock(BeanFactory.class)); + ThreadPoolTaskScheduler sched = new ThreadPoolTaskScheduler(); + endpoint.setTaskScheduler(sched); + endpoint.setConfirmTimeout(100); + sched.afterPropertiesSet(); + endpoint.setTaskScheduler(sched); + endpoint.afterPropertiesSet(); + endpoint.start(); + endpoint.handleMessage(message); + verify(template).send(isNull(), isNull(), any(), correlationCaptor.capture()); + CorrelationData correlation = correlationCaptor.getValue(); + correlationList.add(correlation); + assertThat(TestUtils.getPropertyValue(correlation, "message", Message.class)).isSameAs(message); + Message nack = nacks.receive(10_000); + assertThat(nack).isNotNull(); + assertThat(nack.getPayload()).isInstanceOf(NackedAmqpMessageException.class); + assertThat(((NackedAmqpMessageException) nack.getPayload()).getFailedMessage()).isSameAs(message); + assertThat(((NackedAmqpMessageException) nack.getPayload()).getCorrelationData()) + .isSameAs(message.getHeaders().getId()); + assertThat(((NackedAmqpMessageException) nack.getPayload()).getNackReason()).isEqualTo("Confirm timed out"); + endpoint.stop(); + sched.destroy(); + } + @Test public void adapterWithReturns() throws Exception { this.withReturns.setErrorMessageStrategy(null); @@ -162,7 +214,7 @@ public class AmqpOutboundEndpointTests { } @Test - public void adapterWithReturnsAndErrorMessageStrategy() throws Exception { + public void adapterWithReturnsAndErrorMessageStrategy() { Message message = MessageBuilder.withPayload("hello").build(); this.returnRequestChannel.send(message); Message returned = returnChannel.receive(10000); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java index 6b51267a9d..4144d10e33 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java @@ -106,7 +106,7 @@ public class AsyncAmqpGatewayTests { try { waitForAckBeforeReplying.await(10, TimeUnit.SECONDS); } - catch (InterruptedException e) { + catch (@SuppressWarnings("unused") InterruptedException e) { Thread.currentThread().interrupt(); } return foo.toUpperCase(); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 0436b3b7c3..eaba227eba 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -491,6 +491,15 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann ---- ==== +[[amqp-outbound-endpoints]] +=== Outbound Channel Adapter + +The following outbound endpoints have many similar configuration options. +Starting with version 5.2, the `confirm-timeout` has been added. +Normally, when publisher confirms are enabled, the broker will quickly return an ack (or nack) which will be sent to the appropriate channel. +If a channel is closed before the confirm is received, the Spring AMQP framework will synthesize a nack. +"Missing" acks should never occur but, if you set this property, the endpoint will periodically check for them and synthesize a nack if the time elapses without a confirm being received. + [[amqp-outbound-channel-adapter]] === Outbound Channel Adapter @@ -511,11 +520,12 @@ The following example shows the available properties for an AMQP outbound channe confirm-correlation-expression="" <10> confirm-ack-channel="" <11> confirm-nack-channel="" <12> - return-channel="" <13> - error-message-strategy="" <14> - header-mapper="" <15> - mapped-request-headers="" <16> - lazy-connect="true" /> <17> + confirm-timeout="" <13> + return-channel="" <14> + error-message-strategy="" <15> + header-mapper="" <16> + mapped-request-headers="" <17> + lazy-connect="true" /> <18> ---- @@ -570,21 +580,24 @@ The payload is the correlation data defined by the `confirm-correlation-expressi If the expression is `#root` or `#this`, the message is built from the original message, with the `amqp_publishConfirm` header set to `false`. When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `NackedAmqpMessageException` payload. Optional (the default is `nullChannel`). -<13> The channel to which returned messages are sent. +<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. 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. -<14> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages. -<15> A reference to an `AmqpHeaderMapper` to use when sending AMQP Messages. +<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. 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. -<16> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message. +<17> 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"`). -<17> When set to `false`, the endpoint attempts to connect to the broker during application context initialization. +<18> 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. ==== @@ -691,7 +704,7 @@ The following listing shows the possible properties for an AMQP Outbound Gateway ==== [source,xml] ---- - + request-channel="myRequestChannel" <2> amqp-template="" <3> exchange-name="" <4> @@ -706,9 +719,10 @@ The following listing shows the possible properties for an AMQP Outbound Gateway confirm-correlation-expression="" <13> confirm-ack-channel="" <14> confirm-nack-channel="" <15> - return-channel="" <16> - error-message-strategy="" <17> - lazy-connect="true" /> <18> + confirm-timeout="" <16> + return-channel="" <17> + error-message-strategy="" <18> + lazy-connect="true" /> <19> ---- @@ -771,13 +785,16 @@ The payload is the correlation data defined by `confirm-correlation-expression` If the expression is `#root` or `#this`, the message is built from the original message, with the `amqp_publishConfirm` header set to `false`. When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `NackedAmqpMessageException` payload. Optional (the default is `nullChannel`). -<16> The channel to which returned messages are sent. +<16> When set, the gateway 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). +<17> 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`, and `amqp_returnRoutingKey`. When there is an `ErrorMessageStrategy`, the message is an `ErrorMessage` with a `ReturnedAmqpMessageException` payload. Optional. -<17> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages. -<18> When set to `false`, the endpoint attempts to connect to the broker during application context initialization. +<18> A reference to an `ErrorMessageStrategy` implementation used to build `ErrorMessage` instances when sending returned or negatively acknowledged messages. +<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 by logging an error message 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. ==== @@ -901,7 +918,7 @@ The following listing shows the possible configuration options for an AMQP async ==== [source,xml] ---- - + request-channel="myRequestChannel" <2> async-template="" <3> exchange-name="" <4> @@ -916,8 +933,9 @@ The following listing shows the possible configuration options for an AMQP async confirm-correlation-expression="" <13> confirm-ack-channel="" <14> confirm-nack-channel="" <15> - return-channel="" <16> - lazy-connect="true" /> <17> + confirm-timeout="" <16> + return-channel="" <17> + lazy-connect="true" /> <18> ---- @@ -978,12 +996,15 @@ The channel to which negative (`nack`) publisher confirmations are sent. The payload is the correlation data defined by the `confirm-correlation-expression`. Requires the underlying `AsyncRabbitTemplate` to have its `enableConfirms` property set to `true`. Optional (the default is `nullChannel`). -<16> The channel to which returned messages are sent. +<16> When set, the gateway 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). +<17> The channel to which returned messages are sent. When provided, the underlying AMQP template is configured to return undeliverable messages to the gateway. The message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, and `amqp_returnRoutingKey`. Requires the underlying `AsyncRabbitTemplate` to have its `mandatory` property set to `true`. Optional. -<17> When set to `false`, the endpoint tries to connect to the broker during application context initialization. +<18> When set to `false`, the endpoint tries to connect to the broker during application context initialization. Doing so allows "`fail fast`" detection of bad configuration, by logging an error message 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 ad87b341e2..d424efe1c8 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -16,6 +16,12 @@ See <> for more information. [[x5.2-general]] === General Changes +[[x5.2-amqp]] +==== AMQP Changes + +The outbound endpoints can now be configured to synthesize a "nack" if no publisher confirm is received within a timeout. +See <> for more information. + [[x5.2-file]] ==== File Changes