INT-3791: Don't Wrap Confirms if Already Message

JIRA: https://jira.spring.io/browse/INT-3791

Do not wrap a publisher confirm if the correlation data is a Messge<?>.

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

@@ -405,7 +405,11 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
 			headers.put(AmqpHeaders.PUBLISH_CONFIRM_NACK_CAUSE, cause);
 		}
 
-		Message<Object> confirmMessage = this.getMessageBuilderFactory().withPayload(userCorrelationData)
+		AbstractIntegrationMessageBuilder<?> builder = userCorrelationData instanceof Message
+				? this.getMessageBuilderFactory().fromMessage((Message<?>) userCorrelationData)
+				: this.getMessageBuilderFactory().withPayload(userCorrelationData);
+
+		Message<?> confirmMessage = builder
 				.copyHeaders(headers)
 				.build();
 		if (ack && this.confirmAckChannel != null) {

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

@@ -147,7 +147,8 @@ public class AmqpInboundGatewayParserTests {
 	@Test
 	public void testInt2971HeaderMapperAndMappedHeadersExclusivity() {
 		try {
-			new ClassPathXmlApplicationContext("AmqpInboundGatewayParserTests-headerMapper-fail-context.xml", this.getClass());
+			new ClassPathXmlApplicationContext("AmqpInboundGatewayParserTests-headerMapper-fail-context.xml",
+					this.getClass()).close();
 		}
 		catch (BeanDefinitionParsingException e) {
 			assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " +

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

@@ -93,6 +93,19 @@
 
 	<int:channel id="pcRequestChannel"/>
 
+	<rabbit:template id="amqpTemplateConfirmsMC" connection-factory="connectionFactory" reply-timeout="10"/>
+
+	<amqp:outbound-gateway request-channel="pcMessageCorrelationRequestChannel"
+						   reply-channel="fromRabbit"
+						   exchange-name="si.test.exchange"
+						   mapped-request-headers="foo*"
+						   requires-reply="false"
+						   amqp-template="amqpTemplateConfirmsMC"
+						   confirm-correlation-expression="#this"
+						   confirm-ack-channel="ackChannel"/>
+
+	<int:channel id="pcMessageCorrelationRequestChannel"/>
+
 	<int:channel id="ackChannel">
 		<int:queue/>
 	</int:channel>

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

@@ -339,7 +339,7 @@ public class AmqpOutboundGatewayParserTests {
 	public void testInt2971HeaderMapperAndMappedHeadersExclusivity() {
 		try {
 			new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-headerMapper-fail-context.xml",
-					this.getClass());
+					this.getClass()).close();;
 		}
 		catch (BeanDefinitionParsingException e) {
 			assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " +
@@ -371,6 +371,19 @@ public class AmqpOutboundGatewayParserTests {
 		assertNotNull(ack);
 		assertEquals("foo", ack.getPayload());
 		assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
+
+		// test whole message is correlation
+		requestChannel = context.getBean("pcMessageCorrelationRequestChannel", MessageChannel.class);
+		message = MessageBuilder.withPayload("hello")
+				.build();
+		requestChannel.send(message);
+		publisherCallbackChannel.handleAck(0, false);
+		ack = ackChannel.receive(1000);
+		assertNotNull(ack);
+		assertSame(message.getPayload(), ack.getPayload());
+		assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
+
+		context.close();
 	}
 
 	public static class FooAdvice extends AbstractRequestHandlerAdvice {

src/reference/asciidoc/amqp.adoc

@@ -7,24 +7,24 @@
 Spring Integration provides Channel Adapters for receiving and sending messages using the Advanced Message Queuing Protocol (AMQP).
 The following adapters are available:
 
-* Inbound Channel Adapter
-* Outbound Channel Adapter
-* Inbound Gateway
-* Outbound Gateway
+* <<amqp-inbound-channel-adapter,Inbound Channel Adapter>>
+* <<amqp-inbound-gateway,Inbound Gateway>>
+* <<amqp-outbound-channel-adapter,Outbound Channel Adapter>>
+* <<amqp-outbound-gateway,Outbound Gateway>>
 
 Spring Integration also provides a point-to-point Message Channel as well as a publish/subscribe Message Channel backed by AMQP Exchanges and Queues.
 
-In order to provide AMQP support, Spring Integration relies on Spring AMQP (http://www.springsource.org/spring-amqp[http://www.springsource.org/spring-amqp]) which "applies core Spring concepts to the development of AMQP-based messaging solutions".
-Spring AMQP provides similar semantics as Spring JMS (http://static.springsource.org/spring/docs/current/spring-framework-reference/html/jms.html[http://static.springsource.org/spring/docs/current/spring-framework-reference/html/jms.html]).
+In order to provide AMQP support, Spring Integration relies on (http://projects.spring.io/spring-amqp[Spring AMQP])
+which "applies core Spring concepts to the development of AMQP-based messaging solutions".
+Spring AMQP provides similar semantics to (http://docs.spring.io/spring/docs/current/spring-framework-reference/html/jms.html[Spring JMS]).
 
 Whereas the provided AMQP Channel Adapters are intended for unidirectional Messaging (send or receive) only, Spring Integration also provides inbound and outbound AMQP Gateways for request/reply operations.
 
 [TIP]
 =====
-Please familiarize yourself with the reference documentation of the Spring AMQP project as well.
+Please familiarize yourself with the
+http://docs.spring.io/spring-amqp/reference/html/[reference documentation of the Spring AMQP project as well].
 It provides much more in-depth information regarding Spring's integration with AMQP in general and RabbitMQ in particular.
-
-You can find the documentation at: http://static.springsource.org/spring-amqp/reference/html/[http://static.springsource.org/spring-amqp/reference/html/]
 =====
 [[amqp-inbound-channel-adapter]]
 === Inbound Channel Adapter
@@ -385,11 +385,13 @@ Requires a dedicated `RabbitTemplate` and a `CachingConnectionFactory` with the
 `true`. When a publisher confirm is received, and correlation data is supplied, it is written to either the
 confirm-ack-channel, or the confirm-nack-channel, depending on the confirmation type. The payload of the confirm is
 the correlation data as defined by this expression and the message will have a header 'amqp_publishConfirm' set to true (ack) or false (nack).
-Examples: "headers['myCorrelationData']", "payload".
-_Optional_.
+Examples: "`headers['myCorrelationData']`", "`payload`".
 Starting with _version 4.1_ the `amqp_publishConfirmNackCause` message header has been added.
 It contains the `cause` of a 'nack' for publisher confirms.
-
+Starting with _version 4.2_, if the expression resolves to a `Message<?>` instance (such as "`#this`"), the message
+emitted on the ack/nack channel is based on that message, with the additional header(s) added.
+Previously, a new message was created with the correlation data as its payload, regardless of type.
+_Optional_.
 
 <11> The channel to which positive (ack) publisher confirms are sent; payload is the correlation data defined by the _confirm-correlation-expression_.
 _Optional, default=nullChannel_.
@@ -407,11 +409,11 @@ _Optional_.
 
 
 <14> A reference to an `AmqpHeaderMapper` to use when sending AMQP Messages.
-_Optional_.
 By default only standard AMQP properties (e.g.
 `contentType`) will be copied to the Spring Integration `MessageHeaders`.
 Any user-defined headers will NOT be copied to the Message by the default`DefaultAmqpHeaderMapper`.
 Not allowed if 'request-header-names' is provided.
+_Optional_.
 
 
 <15> Comma-separated list of names of AMQP Headers to be mapped from the `MessageHeaders` to the AMQP Message.
@@ -522,10 +524,12 @@ Requires a dedicated `RabbitTemplate` and a `CachingConnectionFactory` with the
 `true`. When a publisher confirm is received, and correlation data is supplied, it is written to either the
 confirm-ack-channel, or the confirm-nack-channel, depending on the confirmation type. The payload of the confirm is
 the correlation data as defined by this expression and the message will have a header 'amqp_publishConfirm' set to true (ack) or false (nack).
+For nacks, an additional header `amqp_publishConfirmNackCause` is provided.
 Examples: "headers['myCorrelationData']", "payload".
+If the expression resolves to a `Message<?>` instance (such as "`#this`"), the message
+emitted on the ack/nack channel is based on that message, with the additional header(s) added.
+Previously, a new message was created with the correlation data as its payload, regardless of type.
 _Optional_.
-Starting with _version 4.1_ the `amqp_publishConfirmNackCause` message header has been added.
-It contains the `cause` of a 'nack' for publisher confirms.
 
 <14> Since _version 4.2_. The channel to which positive (ack) publisher confirms are sent; payload is the correlation data defined by the _confirm-correlation-expression_.
 _Optional, default=nullChannel_.

src/reference/asciidoc/whats-new.adoc

@@ -201,9 +201,15 @@ See <<conditional-pollers>> for more information.
 [[x4.2-amqp-changes]]
 ==== AMQP Changes
 
+===== Publisher Confirms
+
 The `<int-amqp:outbound-gateway>` now supports `confirm-correlation-expression` and `confirm-(n)ack-channel`
 attributes with similar purpose as for `<int-amqp:outbound-channel-adapter>`.
 
+For both the outbound channel adapter and gateway, if the correlation data is a `Message<?>`, it will be the basis
+of the message on the ack/nack channel, with the additional header(s) added.
+Previously, any correlation data (including `Message<?>`) was returned as the payload of the ack/nack message.
+
 See <<amqp>> for more information.
 
 [[x4.2-xpath-splitter]]