INT-4371: Polishing - raw message header; docs

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

* Polishing - DSL

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java

@@ -65,4 +65,9 @@ public class AmqpInboundPolledChannelAdapterSpec
 		return this;
 	}
 
+	public AmqpInboundPolledChannelAdapterSpec rawMessageHeader(boolean rawMessageHeader) {
+		this.target.setRawMessageHeader(rawMessageHeader);
+		return this;
+	}
+
 }

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

@@ -32,6 +32,7 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.context.MessageSource;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.endpoint.AbstractMessageSource;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
@@ -66,6 +67,8 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
 
 	private MessageConverter messageConverter = new SimpleMessageConverter();
 
+	private boolean rawMessageHeader;
+
 	public AmqpMessageSource(ConnectionFactory connectionFactory, String queue) {
 		this(connectionFactory, new AmqpAckCallbackFactory(), queue);
 	}
@@ -132,6 +135,20 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
 		this.messageConverter = messageConverter;
 	}
 
+	protected boolean isRawMessageHeader() {
+		return this.rawMessageHeader;
+	}
+
+	/**
+	 * 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.
+	 */
+	public void setRawMessageHeader(boolean rawMessageHeader) {
+		this.rawMessageHeader = rawMessageHeader;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "amqp:message-source";
@@ -153,11 +170,15 @@ public class AmqpMessageSource extends AbstractMessageSource<Object> {
 			MessageProperties messageProperties = this.propertiesConverter.toMessageProperties(resp.getProps(),
 					resp.getEnvelope(), StandardCharsets.UTF_8.name());
 			Map<String, Object> headers = this.headerMapper.toHeadersFromRequest(messageProperties);
-			Object payload = this.messageConverter
-					.fromMessage(new org.springframework.amqp.core.Message(resp.getBody(), messageProperties));
-			return getMessageBuilderFactory().withPayload(payload)
+			org.springframework.amqp.core.Message amqpMessage = new org.springframework.amqp.core.Message(resp.getBody(), messageProperties);
+			Object payload = this.messageConverter.fromMessage(amqpMessage);
+			AbstractIntegrationMessageBuilder<Object> builder = getMessageBuilderFactory().withPayload(payload)
 					.copyHeaders(headers)
 					.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, callback);
+			if (this.rawMessageHeader) {
+				builder.setHeader(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, amqpMessage);
+			}
+			return builder;
 		}
 		catch (IOException e) {
 			RabbitUtils.closeChannel(channel);

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

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.amqp.inbound;
 
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.willReturn;
@@ -30,6 +32,7 @@ import java.util.concurrent.TimeoutException;
 import org.junit.Test;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
 import org.springframework.integration.support.AcknowledgmentCallback.Status;
 import org.springframework.integration.support.StaticMessageHeaderAccessor;
 import org.springframework.messaging.Message;
@@ -65,7 +68,10 @@ public class AmqpMessageSourceTests {
 
 		CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
 		AmqpMessageSource source = new AmqpMessageSource(ccf, "foo");
+		source.setRawMessageHeader(true);
 		Message<?> received = source.receive();
+		assertThat(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE),
+				instanceOf(org.springframework.amqp.core.Message.class));
 		// 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"

src/reference/asciidoc/amqp.adoc

@@ -301,6 +301,38 @@ public class AmqpJavaApplication {
 }
 ----
 
+=== Polled Inbound Channel Adapter
+
+Starting with _version 5.0.1_, a polled channel adapter is provided, allowing fetching individual messages on-demand, for example with a `MessageSourcePollingTemplate` or a poller.
+See <<deferred-acks-message-source>> for more information.
+
+It does not currently have XML configuration.
+
+[source, java]
+----
+@Bean
+public AmqpMessageSource source(ConnectionFactory connectionFactory) {
+    return new AmpqpMessageSource(connectionFactory, "someQueue");
+}
+----
+
+Refer to the javadocs for configuration properties.
+
+With the Java DSL:
+
+[source, java]
+----
+@Bean
+public IntegrationFlow flow() {
+    return IntegrationFlows.from(Amqp.inboundPolledAdapter(connectionFactory(), DSL_QUEUE),
+                    e -> e.poller(Pollers.fixedDelay(1_000)).autoStartup(false))
+            .handle(p -> {
+                ...
+            })
+            .get();
+}
+----
+
 [[amqp-inbound-gateway]]
 === Inbound Gateway