GH-2797: AMQP: Add confirm-timeout

Resolves #2797

Avoid runtime casting to `RabbitTemplate`.

Author: garyrussell

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 @@ protected AbstractBeanDefinition parseConsumer(Element element, ParserContext pa
 		}
 		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",

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 @@ protected BeanDefinitionBuilder parseHandler(Element element, ParserContext pars
 
 		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

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.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 void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) {
 		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 @@ protected boolean isHeadersMappedLast() {
 		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 @@ protected final void doInit() {
 		}
 		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 synchronized void start() {
 				}
 			}
 			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<CorrelationData> 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 @@ protected AbstractIntegrationMessageBuilder<?> buildReply(MessageConverter conve
 	}
 
 	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 @@ protected Message<?> buildReturnedMessage(org.springframework.amqp.core.Message
 					.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;
 	}

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 String getComponentType() {
 		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 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 
 	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 @@ private void send(String exchangeName, String routingKey,
 
 	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 void confirm(CorrelationData correlationData, boolean ack, String cause)
 	@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);

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.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 String getComponentType() {
 		return "amqp:outbound-async-gateway";
 	}
 
+	@Override
+	protected RabbitTemplate getRabbitTemplate() {
+		return  this.template.getRabbitTemplate();
+	}
+
 	@Override
 	protected void doStart() {
 		super.doStart();

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.
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="confirm-timeout" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+			Generate a negative acknowledgment (nack) if a publisher confirm is not received within this time
+			in milliseconds. Default none (nacks will not be generated).
+							]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="error-message-strategy" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml

@@ -54,6 +54,19 @@
 								   confirm-ack-channel="ackChannel"
 								   error-message-strategy="ems"/>
 
+	<rabbit:template id="amqpTemplateConfirms2" connection-factory="connectionFactory"/>
+
+	<amqp:outbound-channel-adapter id="withPublisherConfirms2" channel="pcRequestChannel"
+								   exchange-name="outboundchanneladapter.test.1"
+								   mapped-request-headers="foo*"
+								   auto-startup="false"
+								   amqp-template="amqpTemplateConfirms2"
+								   confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
+								   confirm-ack-channel="ackChannel"
+								   confirm-nack-channel="nackChannel"
+								   confirm-timeout="2000"
+								   error-message-strategy="ems"/>
+
 	<bean id="ems" class="org.springframework.integration.support.DefaultErrorMessageStrategy" />
 
 	<int:channel id="pcRequestChannel"/>
@@ -62,6 +75,10 @@
 		<int:queue/>
 	</int:channel>
 
+	<int:channel id="nackChannel">
+		<int:queue/>
+	</int:channel>
+
 	<amqp:outbound-channel-adapter id="withDefaultAmqpTemplateExchangeAndRoutingKey"/>
 
 	<rabbit:template id="amqpTemplateWithSuppliedExchangeAndRoutingKey" connection-factory="connectionFactory"

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java

@@ -31,6 +31,7 @@
 
 import java.io.IOException;
 import java.lang.reflect.Field;
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -125,9 +126,9 @@ public void withHeaderMapperCustomHeaders() {
 				.getExpressionString()).isEqualTo("42");
 		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);
 		final AtomicBoolean shouldBePersistent = new AtomicBoolean();
 
@@ -179,6 +180,19 @@ public void parseWithPublisherConfirms() {
 		assertThat(TestUtils.getPropertyValue(endpoint, "errorMessageStrategy")).isSameAs(context.getBean("ems"));
 	}
 
+	@Test
+	public void parseWithPublisherConfirms2() {
+		Object eventDrivenConsumer = context.getBean("withPublisherConfirms2");
+		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
+				AmqpOutboundEndpoint.class);
+		MessageChannel nackChannel = context.getBean("nackChannel", MessageChannel.class);
+		MessageChannel ackChannel = context.getBean("ackChannel", MessageChannel.class);
+		assertThat(TestUtils.getPropertyValue(endpoint, "confirmAckChannel")).isSameAs(ackChannel);
+		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"));
+	}
+
 	@SuppressWarnings("rawtypes")
 	@Test
 	public void amqpOutboundChannelAdapterWithinChain() {
@@ -189,9 +203,9 @@ public void amqpOutboundChannelAdapterWithinChain() {
 		AmqpOutboundEndpoint endpoint = (AmqpOutboundEndpoint) chainHandlers.get(0);
 		assertThat(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode")).isNull();
 
-		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 -> {