Add CorrelationData spring-messaging header

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

* Get header once only.

* Close test connection factories.

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java

@@ -89,6 +89,12 @@ public abstract class AmqpHeaders {
 
 	public static final String SPRING_REPLY_TO_STACK = PREFIX + "springReplyToStack";
 
+	/**
+	 * A CorrelationData instance for publisher confirms (not mapped).
+	 * @since 2.3
+	 */
+	public static final String PUBLISH_CONFIRM_CORRELATION = PREFIX + "publishConfirmCorrelation";
+
 	public static final String PUBLISH_CONFIRM = PREFIX + "publishConfirm";
 
 	public static final String PUBLISH_CONFIRM_NACK_CAUSE = PREFIX + "publishConfirmNackCause";

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -22,11 +22,14 @@
 import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
- * Base class for correlating publisher confirms to sent messages.
- * Use the {@link org.springframework.amqp.rabbit.core.RabbitTemplate}
- * methods that include one of
- * these as a parameter; when the publisher confirm is received,
- * the CorrelationData is returned with the ack/nack.
+ * Base class for correlating publisher confirms to sent messages. Use the
+ * {@link org.springframework.amqp.rabbit.core.RabbitTemplate} methods that include one of
+ * these as a parameter; when the publisher confirm is received, the CorrelationData is
+ * returned with the ack/nack. When returns are also enabled, the
+ * {@link #setReturnedMessage(Message) returnedMessage} property will be populated when a
+ * message can't be delivered - the return always arrives before the confirmation. In this
+ * case the {@code #id} property must be set to a unique value.
+ *
  * @author Gary Russell
  * @since 1.0.1
  *

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitMessagingTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -19,6 +19,8 @@
 import java.util.Map;
 
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.MessagingMessageConverter;
 import org.springframework.beans.factory.InitializingBean;
@@ -195,7 +197,13 @@ public <T> T convertSendAndReceive(String exchange, String routingKey, Object re
 	@Override
 	protected void doSend(String destination, Message<?> message) {
 		try {
-			this.rabbitTemplate.send(destination, createMessage(message));
+			Object correlation = message.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION);
+			if (correlation instanceof CorrelationData) {
+				this.rabbitTemplate.send(destination, createMessage(message), (CorrelationData) correlation);
+			}
+			else {
+				this.rabbitTemplate.send(destination, createMessage(message));
+			}
 		}
 		catch (RuntimeException ex) {
 			throw convertAmqpException(ex);
@@ -204,7 +212,13 @@ protected void doSend(String destination, Message<?> message) {
 
 	protected void doSend(String exchange, String routingKey, Message<?> message) {
 		try {
-			this.rabbitTemplate.send(exchange, routingKey, createMessage(message));
+			Object correlation = message.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION);
+			if (correlation instanceof CorrelationData) {
+				this.rabbitTemplate.send(exchange, routingKey, createMessage(message), (CorrelationData) correlation);
+			}
+			else {
+				this.rabbitTemplate.send(exchange, routingKey, createMessage(message));
+			}
 		}
 		catch (RuntimeException ex) {
 			throw convertAmqpException(ex);

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitOperations.java

@@ -108,6 +108,21 @@ <T> T invoke(OperationsCallback<T> action, @Nullable com.rabbitmq.client.Confirm
 	 */
 	ConnectionFactory getConnectionFactory();
 
+	/**
+	 * Send a message to the default exchange with a specific routing key.
+	 *
+	 * @param routingKey the routing key
+	 * @param message a message to send
+	 * @param correlationData data to correlate publisher confirms.
+	 * @throws AmqpException if there is a problem
+	 * @since 2.3
+	 */
+	default void send(String routingKey, Message message, CorrelationData correlationData)
+			throws AmqpException {
+
+		throw new UnsupportedOperationException("This implementation does not support this method");
+	}
+
 	/**
 	 * Send a message to a specific exchange with a specific routing key.
 	 *

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -1032,6 +1032,11 @@ public void send(String routingKey, Message message) throws AmqpException {
 		send(this.exchange, routingKey, message);
 	}
 
+	@Override
+	public void send(String routingKey, Message message, CorrelationData correlationData) throws AmqpException {
+		send(this.exchange, routingKey, message, correlationData);
+	}
+
 	@Override
 	public void send(final String exchange, final String routingKey, final Message message) throws AmqpException {
 		send(exchange, routingKey, message, null);
@@ -1045,8 +1050,7 @@ public void send(final String exchange, final String routingKey,
 			doSend(channel, exchange, routingKey, message,
 					(RabbitTemplate.this.returnsCallback != null
 							|| (correlationData != null && StringUtils.hasText(correlationData.getId())))
-							&& RabbitTemplate.this.mandatoryExpression.getValue(
-							RabbitTemplate.this.evaluationContext, message, Boolean.class),
+							&& isMandatoryFor(message),
 					correlationData);
 			return null;
 		}, obtainTargetConnectionFactory(this.sendConnectionFactorySelectorExpression, message));

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 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.
+ * 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.amqp.rabbit.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
+import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.messaging.support.GenericMessage;
+
+/**
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+@RabbitAvailable("messaging.confirms")
+public class MessagingTemplateConfirmsTests {
+
+	@Test
+	void confirmHeader() throws Exception {
+		CachingConnectionFactory ccf = new CachingConnectionFactory(
+				RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
+		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
+		RabbitTemplate rt = new RabbitTemplate(ccf);
+		RabbitMessagingTemplate rmt = new RabbitMessagingTemplate(rt);
+		CorrelationData data = new CorrelationData();
+		rmt.send("messaging.confirms",
+				new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data)));
+		assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
+		ccf.destroy();
+	}
+
+	@Test
+	void confirmHeaderUnroutable() throws Exception {
+		CachingConnectionFactory ccf = new CachingConnectionFactory(
+				RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
+		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
+		ccf.setPublisherReturns(true);
+		RabbitTemplate rt = new RabbitTemplate(ccf);
+		rt.setMandatory(true);
+		RabbitMessagingTemplate rmt = new RabbitMessagingTemplate(rt);
+		CorrelationData data = new CorrelationData("foo");
+		rmt.send("messaging.confirms.unroutable",
+				new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data)));
+		assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
+		assertThat(data.getReturnedMessage()).isNotNull();
+		ccf.destroy();
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitMessagingTemplateTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -28,6 +28,7 @@
 import static org.mockito.Mockito.verify;
 
 import java.io.Writer;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -40,14 +41,17 @@
 
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.test.MessageTestUtils;
+import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.MessagingMessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.GenericMessageConverter;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 
 
@@ -83,7 +87,8 @@ public void verifyConverter() {
 		RabbitTemplate template = new RabbitTemplate(mock(ConnectionFactory.class));
 		RabbitMessagingTemplate rmt = new RabbitMessagingTemplate(template);
 		rmt.afterPropertiesSet();
-		assertThat(TestUtils.getPropertyValue(rmt, "amqpMessageConverter.payloadConverter")).isSameAs(template.getMessageConverter());
+		assertThat(TestUtils.getPropertyValue(rmt, "amqpMessageConverter.payloadConverter"))
+				.isSameAs(template.getMessageConverter());
 
 		rmt = new RabbitMessagingTemplate(template);
 		MessagingMessageConverter amqpMessageConverter = new MessagingMessageConverter();
@@ -94,6 +99,20 @@ public void verifyConverter() {
 		assertThat(TestUtils.getPropertyValue(rmt, "amqpMessageConverter.payloadConverter")).isSameAs(payloadConverter);
 	}
 
+	@Test
+	void correlation() {
+		this.messagingTemplate.setDefaultDestination("defRk");
+		this.messagingTemplate.send(new GenericMessage<>("foo",
+				Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, new CorrelationData())));
+		verify(this.rabbitTemplate).send(eq("defRk"), any(), any(CorrelationData.class));
+		this.messagingTemplate.send("rk", new GenericMessage<>("foo",
+				Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, new CorrelationData())));
+		verify(this.rabbitTemplate).send(eq("rk"), any(), any(CorrelationData.class));
+		this.messagingTemplate.send("ex", "rk", new GenericMessage<>("foo",
+				Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, new CorrelationData())));
+		verify(this.rabbitTemplate).send(eq("ex"), eq("rk"), any(), any(CorrelationData.class));
+	}
+
 	@Test
 	public void send() {
 		Message<String> message = createTextMessage();