GH-2759: Fix CorrelationData.future

* GH-2759: Fix CorrelationData.future

Fixes #2759

The outbound endpoints wrap user correlation data in a wrapper.
If the user data is a `CorrelationData`, we must delegate methods
involving the `Future<?>` and `returnedMessage` to the user data.

**cherry-pick to 5.1 and switch AMQP to snapshots**

* Polishing - remove redundant override.

* Add debug log with null correlation data

# Conflicts:
#	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java

* Updated Spring AMQP dependency to `2.1.5.BUILD-SNAPSHOT`
* Moved `AmqpOutboundEndpointTests` assertions to `AssertJ` to avoid
conflicts with `master`

Author: garyrussell

build.gradle

@@ -134,7 +134,7 @@ subprojects { subproject ->
 		romeToolsVersion = '1.9.0'
 		servletApiVersion = '4.0.0'
 		smackVersion = '4.3.1'
-		springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.1.4.RELEASE'
+		springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.1.5.BUILD-SNAPSHOT'
 		springDataJpaVersion = '2.1.5.RELEASE'
 		springDataMongoVersion = '2.1.5.RELEASE'
 		springDataRedisVersion = '2.1.5.RELEASE'

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

@@ -44,6 +44,7 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
  * @author Gary Russell
@@ -479,8 +480,14 @@ protected CorrelationData generateCorrelationData(Message<?> requestMessage) {
 			if (messageId == null) {
 				messageId = NO_ID;
 			}
-			correlationData = new CorrelationDataWrapper(messageId.toString(),
-					this.correlationDataGenerator.processMessage(requestMessage), requestMessage);
+			Object userData = this.correlationDataGenerator.processMessage(requestMessage);
+			if (userData != null) {
+				correlationData = new CorrelationDataWrapper(messageId.toString(), userData, requestMessage);
+			}
+			else {
+				this.logger.debug("'confirmCorrelationExpression' resolved to 'null'; "
+						+ "no publisher confirm will be sent to the ack or nack channel");
+			}
 		}
 		return correlationData;
 	}
@@ -604,6 +611,23 @@ public Message<?> getMessage() {
 			return this.message;
 		}
 
+		@Override
+		public SettableListenableFuture<Confirm> getFuture() {
+			if (this.userData instanceof CorrelationData) {
+				return ((CorrelationData) this.userData).getFuture();
+			}
+			else {
+				return super.getFuture();
+			}
+		}
+
+		@Override
+		public void setReturnedMessage(org.springframework.amqp.core.Message returnedMessage) {
+			if (this.userData instanceof CorrelationData) {
+				((CorrelationData) this.userData).setReturnedMessage(returnedMessage);
+			}
+			super.setReturnedMessage(returnedMessage);
+		}
 	}
 
 }

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml

@@ -70,6 +70,7 @@
 								   routing-key="#{queue.name + queue.name}"
 								   mapped-request-headers="foo*"
 								   amqp-template="amqpTemplateReturns"
+								   confirm-correlation-expression="headers['corrData']"
 								   return-channel="returnChannel" />
 
 	<int:channel id="returnRequestChannel"/>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.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.
@@ -16,18 +16,18 @@
 
 package org.springframework.integration.amqp.outbound;
 
-import static org.hamcrest.Matchers.instanceOf;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+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.junit.BrokerRunning;
 import org.springframework.amqp.support.AmqpHeaders;
@@ -108,24 +108,24 @@ public void testGatewayPublisherConfirms() throws Exception {
 				.build();
 		this.pcRequestChannel.send(message);
 		Message<?> ack = this.ackChannel.receive(10000);
-		assertNotNull(ack);
-		assertEquals("foo", ack.getPayload());
-		assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
+		assertThat(ack).isNotNull();
+		assertThat(ack.getPayload()).isEqualTo("foo");
+		assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE);
 
 		org.springframework.amqp.core.Message received = this.amqpTemplateConfirms.receive(this.queue.getName());
-		assertEquals("\"hello\"", new String(received.getBody(), "UTF-8"));
-		assertEquals("application/json", received.getMessageProperties().getContentType());
-		assertEquals("java.lang.String", received.getMessageProperties().getHeaders()
-				.get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")));
+		assertThat(new String(received.getBody(), "UTF-8")).isEqualTo("\"hello\"");
+		assertThat(received.getMessageProperties().getContentType()).isEqualTo("application/json");
+		assertThat(received.getMessageProperties().getHeaders()
+				.get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))).isEqualTo("java.lang.String");
 
 		// test whole message is correlation
 		message = MessageBuilder.withPayload("hello")
 				.build();
 		this.pcMessageCorrelationRequestChannel.send(message);
 		ack = ackChannel.receive(10000);
-		assertNotNull(ack);
-		assertSame(message.getPayload(), ack.getPayload());
-		assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
+		assertThat(ack).isNotNull();
+		assertThat(ack.getPayload()).isSameAs(message.getPayload());
+		assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE);
 
 		while (this.amqpTemplateConfirms.receive(this.queue.getName()) != null) {
 			// drain
@@ -139,31 +139,38 @@ public void adapterWithPublisherConfirms() throws Exception {
 				.build();
 		this.pcRequestChannelForAdapter.send(message);
 		Message<?> ack = this.ackChannel.receive(10000);
-		assertNotNull(ack);
-		assertEquals("foo", ack.getPayload());
-		assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
+		assertThat(ack).isNotNull();
+		assertThat(ack.getPayload()).isEqualTo("foo");
+		assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE);
 	}
 
 	@Test
 	public void adapterWithReturns() throws Exception {
 		this.withReturns.setErrorMessageStrategy(null);
-		Message<?> message = MessageBuilder.withPayload("hello").build();
+		CorrelationData corrData = new CorrelationData("adapterWithReturns");
+		Message<?> message = MessageBuilder.withPayload("hello")
+				.setHeader("corrData", corrData)
+				.build();
 		this.returnRequestChannel.send(message);
 		Message<?> returned = returnChannel.receive(10000);
-		assertNotNull(returned);
-		assertEquals(message.getPayload(), returned.getPayload());
+		assertThat(returned).isNotNull();
+		assertThat(returned.getPayload()).isEqualTo(message.getPayload());
+		Confirm confirm = corrData.getFuture().get(10, TimeUnit.SECONDS);
+		assertThat(confirm).isNotNull();
+		assertThat(confirm.isAck()).isTrue();
+		assertThat(corrData.getReturnedMessage()).isNotNull();
 	}
 
 	@Test
 	public void adapterWithReturnsAndErrorMessageStrategy() throws Exception {
 		Message<?> message = MessageBuilder.withPayload("hello").build();
 		this.returnRequestChannel.send(message);
 		Message<?> returned = returnChannel.receive(10000);
-		assertNotNull(returned);
-		assertThat(returned, instanceOf(ErrorMessage.class));
-		assertThat(returned.getPayload(), instanceOf(ReturnedAmqpMessageException.class));
+		assertThat(returned).isNotNull();
+		assertThat(returned).isInstanceOf(ErrorMessage.class);
+		assertThat(returned.getPayload()).isInstanceOf(ReturnedAmqpMessageException.class);
 		ReturnedAmqpMessageException payload = (ReturnedAmqpMessageException) returned.getPayload();
-		assertEquals(message.getPayload(), payload.getFailedMessage().getPayload());
+		assertThat(payload.getFailedMessage().getPayload()).isEqualTo(message.getPayload());
 	}
 
 	@Test
@@ -178,18 +185,18 @@ public void adapterWithContentType() throws Exception {
 				.build();
 		this.ctRequestChannel.send(message);
 		org.springframework.amqp.core.Message m = receive(template);
-		assertNotNull(m);
-		assertEquals("\"hello\"", new String(m.getBody(), "UTF-8"));
-		assertEquals("application/json", m.getMessageProperties().getContentType());
-		assertEquals("java.lang.String",
-				m.getMessageProperties().getHeaders().get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")));
+		assertThat(m).isNotNull();
+		assertThat(new String(m.getBody(), "UTF-8")).isEqualTo("\"hello\"");
+		assertThat(m.getMessageProperties().getContentType()).isEqualTo("application/json");
+		assertThat(m.getMessageProperties().getHeaders().get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")))
+				.isEqualTo("java.lang.String");
 		message = MessageBuilder.withPayload("hello")
 				.build();
 		this.ctRequestChannel.send(message);
 		m = receive(template);
-		assertNotNull(m);
-		assertEquals("hello", new String(m.getBody(), "UTF-8"));
-		assertEquals("text/plain", m.getMessageProperties().getContentType());
+		assertThat(m).isNotNull();
+		assertThat(new String(m.getBody(), "UTF-8")).isEqualTo("hello");
+		assertThat(m.getMessageProperties().getContentType()).isEqualTo("text/plain");
 		while (template.receive() != null) {
 			// drain
 		}
@@ -202,7 +209,7 @@ private org.springframework.amqp.core.Message receive(RabbitTemplate template) t
 			Thread.sleep(100);
 			message = template.receive();
 		}
-		assertNotNull(message);
+		assertThat(message).isNotNull();
 		return message;
 	}