AMQP: Add bindSourceMessage property (inbound)

Resolves https://github.com/spring-projects/spring-integration/issues/2958

# Conflicts:
#	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
#	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
#	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java
#	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
#	spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java

@@ -69,6 +69,8 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 
 	private RecoveryCallback<? extends Object> recoveryCallback;
 
+	private boolean bindSourceMessage;
+
 	public AmqpInboundChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
 		Assert.notNull(listenerContainer, "listenerContainer must not be null");
 		Assert.isNull(listenerContainer.getMessageListener(),
@@ -116,6 +118,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 	}
 
 
+	/**
+	 * Set to true to bind the source message in the header named
+	 * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}.
+	 * @param bindSourceMessage true to bind.
+	 * @since 5.1.6
+	 */
+	public void setBindSourceMessage(boolean bindSourceMessage) {
+		this.bindSourceMessage = bindSourceMessage;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "amqp:inbound-channel-adapter";
@@ -249,6 +261,9 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 			if (AmqpInboundChannelAdapter.this.retryTemplate != null) {
 				headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger());
 			}
+			if (AmqpInboundChannelAdapter.this.bindSourceMessage) {
+				headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, message);
+			}
 			final org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
 					.withPayload(payload)
 					.copyHeaders(headers)

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java

@@ -81,6 +81,8 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 
 	private RecoveryCallback<? extends Object> recoveryCallback;
 
+	private boolean bindSourceMessage;
+
 	public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer) {
 		this(listenerContainer, new RabbitTemplate(listenerContainer.getConnectionFactory()), false);
 	}
@@ -175,6 +177,16 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 		this.recoveryCallback = recoveryCallback;
 	}
 
+	/**
+	 * Set to true to bind the source message in the header named
+	 * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}.
+	 * @param bindSourceMessage true to bind.
+	 * @since 5.1.6
+	 */
+	public void setBindSourceMessage(boolean bindSourceMessage) {
+		this.bindSourceMessage = bindSourceMessage;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "amqp:inbound-gateway";
@@ -295,6 +307,9 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 				if (AmqpInboundGateway.this.retryTemplate != null) {
 					headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger());
 				}
+				if (AmqpInboundGateway.this.bindSourceMessage) {
+					headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, message);
+				}
 			}
 			catch (RuntimeException e) {
 				MessageChannel errorChannel = getErrorChannel();

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java

@@ -142,10 +142,11 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
 	}
 
 	/**
-	 * Set to true to include the raw spring-amqp message as a header
-	 * with key {@link AmqpMessageHeaderErrorMessageStrategy#AMQP_RAW_MESSAGE},
-	 * enabling callers to have access to the message to process errors.
-	 * @param rawMessageHeader true to include the header.
+	 * Set to true to include the raw spring-amqp message as a header with key
+	 * {@link AmqpMessageHeaderErrorMessageStrategy#AMQP_RAW_MESSAGE}, enabling callers to
+	 * have access to the message to process errors. The raw message is also added to the
+	 * common header {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}.
+	 * @param rawMessageHeader true to include the headers.
 	 */
 	public void setRawMessageHeader(boolean rawMessageHeader) {
 		this.rawMessageHeader = rawMessageHeader;
@@ -180,6 +181,7 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
 					.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, callback);
 			if (this.rawMessageHeader) {
 				builder.setHeader(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, amqpMessage);
+				builder.setHeader(IntegrationMessageHeaderAccessor.SOURCE_DATA, amqpMessage);
 			}
 			return builder;
 		}

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.amqp.inbound;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.sameInstance;
 import static org.junit.Assert.assertThat;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.isNull;
@@ -32,6 +33,7 @@ import org.junit.Test;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback.Status;
 import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
@@ -46,6 +48,7 @@ import com.rabbitmq.client.GetResponse;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.0.1
  *
@@ -73,6 +76,8 @@ public class AmqpMessageSourceTests {
 		assertThat(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE),
 				instanceOf(org.springframework.amqp.core.Message.class));
 		assertThat(received.getHeaders().get(AmqpHeaders.CONSUMER_QUEUE), equalTo("foo"));
+		assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA),
+				sameInstance(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE)));
 		// make sure channel is not cached
 		org.springframework.amqp.rabbit.connection.Connection conn = ccf.createConnection();
 		Channel notCached = conn.createChannel(false); // should not have been "closed"

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java

@@ -107,6 +107,7 @@ public class InboundEndpointTests {
 
 		adapter.setOutputChannel(channel);
 		adapter.setBeanFactory(mock(BeanFactory.class));
+		adapter.setBindSourceMessage(true);
 		adapter.afterPropertiesSet();
 
 		Object payload = new Foo("bar1");
@@ -129,6 +130,8 @@ public class InboundEndpointTests {
 
 		assertSame(rabbitChannel, result.getHeaders().get(AmqpHeaders.CHANNEL));
 		assertEquals(123L, result.getHeaders().get(AmqpHeaders.DELIVERY_TAG));
+		org.springframework.amqp.core.Message sourceData = StaticMessageHeaderAccessor.getSourceData(result);
+		assertThat(sourceData).isSameAs(amqpMessage);
 	}
 
 	@Test
@@ -162,6 +165,8 @@ public class InboundEndpointTests {
 		Message<?> result = new JsonToObjectTransformer().transform(receive);
 
 		assertEquals(payload, result.getPayload());
+		org.springframework.amqp.core.Message sourceData = StaticMessageHeaderAccessor.getSourceData(result);
+		assertThat(sourceData).isNull();
 	}
 
 	@Test

spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java

@@ -66,6 +66,12 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
 
 	public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback";
 
+	/**
+	 * Raw source message.
+	 */
+	public static final String SOURCE_DATA = "sourceData";
+
+
 	private Set<String> readOnlyHeaders = new HashSet<>();
 
 	public IntegrationMessageHeaderAccessor(@Nullable Message<?> message) {
@@ -149,6 +155,18 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
 		return getHeader(DELIVERY_ATTEMPT, AtomicInteger.class);
 	}
 
+	/**
+	 * Get the source data header, if present.
+	 * @param <T> the data type.
+	 * @return the source header.
+	 * @since 5.1.6
+	 */
+	@SuppressWarnings("unchecked")
+	@Nullable
+	public <T> T getSourceData() {
+		return (T) getHeader(SOURCE_DATA);
+	}
+
 	@SuppressWarnings("unchecked")
 	@Nullable
 	public <T> T getHeader(String key, Class<T> type) {

spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java

@@ -107,4 +107,10 @@ public final class StaticMessageHeaderAccessor {
 				AcknowledgmentCallback.class);
 	}
 
+	@SuppressWarnings("unchecked")
+	@Nullable
+	public static <T> T getSourceData(Message<?> message) {
+		return (T) message.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA);
+	}
+
 }