AMQP: Support CorrelationData Message Headers

In preparation for spring-attic/spring-cloud-stream-binder-rabbit#303

Now that `CorrelationData` has a `Future<?>`, users might simply add
correlation data in a header and not receive confirm/return messages.

- No longer require channels for returns and confirms
- don't build the confirm message if there are no channels
- reduce the log level for no channels to DEBUG
- complete the user's future when a message is returned (async GW)

* Fix typo.

Author: garyrussell

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java

@@ -63,7 +63,7 @@
 public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
 		implements Lifecycle {
 
-	private static final UUID NO_ID = new UUID(0L, 0L);
+	private static final String NO_ID = new UUID(0L, 0L).toString();
 
 	private String exchangeName;
 
@@ -548,20 +548,33 @@ public boolean isRunning() {
 
 	protected CorrelationData generateCorrelationData(Message<?> requestMessage) {
 		CorrelationData correlationData = null;
+		UUID uuid = requestMessage.getHeaders().getId();
+		String messageId;
+		if (uuid == null) {
+			messageId = NO_ID;
+		}
+		else {
+			messageId = uuid.toString();
+		}
 		if (this.correlationDataGenerator != null) {
-			UUID messageId = requestMessage.getHeaders().getId();
-			if (messageId == null) {
-				messageId = NO_ID;
-			}
 			Object userData = this.correlationDataGenerator.processMessage(requestMessage);
 			if (userData != null) {
-				correlationData = new CorrelationDataWrapper(messageId.toString(), userData, requestMessage);
+				correlationData = new CorrelationDataWrapper(messageId, userData, requestMessage);
 			}
 			else {
 				this.logger.debug("'confirmCorrelationExpression' resolved to 'null'; "
 						+ "no publisher confirm will be sent to the ack or nack channel");
 			}
 		}
+		if (correlationData == null) {
+			Object correlation = requestMessage.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION);
+			if (correlation instanceof CorrelationData) {
+				correlationData = (CorrelationData) correlation;
+			}
+			if (correlationData != null) {
+				correlationData = new CorrelationDataWrapper(messageId, correlationData, requestMessage);
+			}
+		}
 		return correlationData;
 	}
 
@@ -652,19 +665,21 @@ protected void handleConfirm(CorrelationData correlationData, boolean ack, Strin
 			return;
 		}
 		Object userCorrelationData = wrapper.getUserData();
-		Message<?> confirmMessage;
-		confirmMessage = buildConfirmMessage(ack, cause, wrapper, userCorrelationData);
-		if (ack && getConfirmAckChannel() != null) {
-			sendOutput(confirmMessage, getConfirmAckChannel(), true);
-		}
-		else if (!ack && getConfirmNackChannel() != null) {
-			sendOutput(confirmMessage, getConfirmNackChannel(), true);
+		MessageChannel ackChannel = getConfirmAckChannel();
+		if (ack && ackChannel != null) {
+			sendOutput(buildConfirmMessage(ack, cause, wrapper, userCorrelationData), ackChannel, true);
 		}
 		else {
-			if (logger.isInfoEnabled()) {
-				logger.info("Nowhere to send publisher confirm "
-						+ (ack ? "ack" : "nack") + " for "
-						+ userCorrelationData);
+			MessageChannel nackChannel = getConfirmNackChannel();
+			if (!ack && nackChannel != null) {
+				sendOutput(buildConfirmMessage(ack, cause, wrapper, userCorrelationData), nackChannel, true);
+			}
+			else {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Nowhere to send publisher confirm "
+							+ (ack ? "ack" : "nack") + " for "
+							+ userCorrelationData);
+				}
 			}
 		}
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java

@@ -22,12 +22,14 @@
 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.connection.CorrelationData.Confirm;
 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;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
@@ -87,20 +89,23 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 		addDelayProperty(requestMessage, amqpMessage);
 		RabbitMessageFuture future = this.template.sendAndReceive(generateExchangeName(requestMessage),
 				generateRoutingKey(requestMessage), amqpMessage);
-		future.addCallback(new FutureCallback(requestMessage));
 		CorrelationData correlationData = generateCorrelationData(requestMessage);
 		if (correlationData != null && future.getConfirm() != null) {
 			future.getConfirm().addCallback(new CorrelationCallback(correlationData, future));
 		}
+		future.addCallback(new FutureCallback(requestMessage, correlationData));
 		return null;
 	}
 
 	private final class FutureCallback implements ListenableFutureCallback<org.springframework.amqp.core.Message> {
 
 		private final Message<?> requestMessage;
 
-		FutureCallback(Message<?> requestMessage) {
+		private final CorrelationDataWrapper correlationData;
+
+		FutureCallback(Message<?> requestMessage, CorrelationData correlationData) {
 			this.requestMessage = requestMessage;
+			this.correlationData = (CorrelationDataWrapper) correlationData;
 		}
 
 		@Override
@@ -141,18 +146,21 @@ public void onFailure(Throwable ex) {
 				}
 			}
 			if (ex instanceof AmqpMessageReturnedException) {
-				if (getReturnChannel() == null) {
-					logger.error("Returned message received and no return channel "
-							+ ((AmqpMessageReturnedException) ex).getReturnedMessage());
-				}
-				else {
-					AmqpMessageReturnedException amre = (AmqpMessageReturnedException) ex;
+				AmqpMessageReturnedException amre = (AmqpMessageReturnedException) ex;
+				MessageChannel returnChannel = getReturnChannel();
+				if (returnChannel != null) {
 					Message<?> returnedMessage = buildReturnedMessage(
 							new ReturnedMessage(amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(),
 									amre.getExchange(), amre.getRoutingKey()),
 							AsyncAmqpOutboundGateway.this.messageConverter);
-					sendOutput(returnedMessage, getReturnChannel(), true);
+					sendOutput(returnedMessage, returnChannel, true);
 				}
+				this.correlationData.setReturnedMessage(amre.getReturnedMessage());
+				/*
+				 *  Complete the user's future (if present) since the async template will only complete
+				 *  once, successfully, or with a failure.
+				 */
+				this.correlationData.getFuture().set(new Confirm(true, null));
 			}
 			else {
 				sendErrorMessage(this.requestMessage, exceptionToSend);

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -20,6 +20,7 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.Collections;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 
@@ -29,16 +30,19 @@
 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.connection.CorrelationData;
 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.amqp.support.AmqpHeaders;
 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.integration.support.MessageBuilder;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
@@ -74,6 +78,17 @@ void testWithReturn(@Autowired IntegrationFlow flow) {
 				.isEqualTo("Message was returned by the broker");
 	}
 
+	@Test
+	void testReturnConfirmNoChannels(@Autowired IntegrationFlow flow2) throws Exception {
+		CorrelationData corr = new CorrelationData("foo");
+		flow2.getInputChannel().send(MessageBuilder.withPayload("test")
+				.setHeader("rk", "junkjunk")
+				.setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, corr)
+				.build());
+		assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
+		assertThat(corr.getReturnedMessage()).isNotNull();
+	}
+
 	@Test
 	@DisabledIf("#{systemEnvironment['TRAVIS'] ?: false}")
 		// needs RabbitMQ 3.7
@@ -106,6 +121,13 @@ public IntegrationFlow flow(RabbitTemplate template) {
 					.waitForConfirm(true));
 		}
 
+		@Bean
+		public IntegrationFlow flow2(RabbitTemplate template) {
+			return f -> f.handle(Amqp.outboundAdapter(template)
+					.exchangeName("")
+					.routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class)));
+		}
+
 		@Bean
 		public CachingConnectionFactory cf() {
 			CachingConnectionFactory ccf = new CachingConnectionFactory(

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -37,6 +37,7 @@
 import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
 import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.junit.RabbitAvailable;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -49,6 +50,7 @@
 import org.springframework.integration.amqp.support.NackedAmqpMessageException;
 import org.springframework.integration.amqp.support.ReturnedAmqpMessageException;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.condition.LogLevels;
@@ -220,4 +222,38 @@ void testConfirmsAndReturns() throws Exception {
 		ccf.destroy();
 	}
 
+	@Test
+	void confirmsAndReturnsNoChannels() throws Exception {
+		CachingConnectionFactory ccf = new CachingConnectionFactory("localhost");
+		ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
+		ccf.setPublisherReturns(true);
+		RabbitTemplate template = new RabbitTemplate(ccf);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(ccf);
+		container.setBeanName("replyContainer");
+		container.setQueueNames("asyncRQ1");
+		container.afterPropertiesSet();
+		container.start();
+		AsyncRabbitTemplate asyncTemplate = new AsyncRabbitTemplate(template, container);
+		asyncTemplate.setEnableConfirms(true);
+		asyncTemplate.setMandatory(true);
+
+		AsyncAmqpOutboundGateway gateway = new AsyncAmqpOutboundGateway(asyncTemplate);
+		gateway.setOutputChannel(new NullChannel());
+		gateway.setExchangeName("");
+		gateway.setRoutingKey("noRoute");
+		gateway.setBeanFactory(mock(BeanFactory.class));
+		gateway.afterPropertiesSet();
+		gateway.start();
+
+		CorrelationData corr = new CorrelationData("foo");
+		gateway.handleMessage(MessageBuilder.withPayload("test")
+				.setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, corr)
+				.build());
+		assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
+		assertThat(corr.getReturnedMessage()).isNotNull();
+
+		asyncTemplate.stop();
+		ccf.destroy();
+	}
+
 }