AMQP: Batch mode option to receive element headers

In preparation for https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/290

SCSt is not set up to receive a `Message<Message<?>>`.

Similar to the kafka endpoint, add the batched headers in a header.

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

@@ -43,6 +43,7 @@ import org.springframework.integration.amqp.support.EndpointUtils;
 import org.springframework.integration.context.OrderlyShutdownCapable;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.support.ErrorMessageUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.RetryOperations;
@@ -65,6 +66,12 @@ import com.rabbitmq.client.Channel;
 public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		OrderlyShutdownCapable {
 
+	/**
+	 * Header containing {@code List<Map<String, Object>} headers when batch mode
+	 * is {@link BatchMode#EXTRACT_PAYLOADS_WITH_HEADERS}.
+	 */
+	public static final String CONSOLIDATED_HEADERS = AmqpHeaders.PREFIX + "batchedHeaders";
+
 	/**
 	 * Defines the payload type when the listener container is configured with consumerBatchEnabled.
 	 */
@@ -80,7 +87,14 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		 * Payload is a {@code List<?>} where each element is the converted body of the
 		 * Spring AMQP Message.
 		 */
-		EXTRACT_PAYLOADS
+		EXTRACT_PAYLOADS,
+
+		/**
+		 * Payload is a {@code List<?>} where each element is the converted body of the
+		 * Spring AMQP Message. The headers for each message are provided in a header
+		 * {@link AmqpInboundChannelAdapter#CONSOLIDATED_HEADERS}.
+		 */
+		EXTRACT_PAYLOADS_WITH_HEADERS
 
 	}
 
@@ -332,7 +346,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 				headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, message);
 			}
 			long deliveryTag = message.getMessageProperties().getDeliveryTag();
-			return createMessageFromPayload(payload, channel, headers, deliveryTag);
+			return createMessageFromPayload(payload, channel, headers, deliveryTag, null);
 		}
 
 		protected Object convertPayload(Message message) {
@@ -350,7 +364,8 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		}
 
 		protected org.springframework.messaging.Message<Object> createMessageFromPayload(Object payload,
-				Channel channel, Map<String, Object> headers, long deliveryTag) {
+				Channel channel, Map<String, Object> headers, long deliveryTag,
+				@Nullable List<Map<String, Object>> listHeaders) {
 
 			if (this.manualAcks) {
 				headers.put(AmqpHeaders.DELIVERY_TAG, deliveryTag);
@@ -359,6 +374,9 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 			if (AmqpInboundChannelAdapter.this.retryTemplate != null) {
 				headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger());
 			}
+			if (listHeaders != null) {
+				headers.put(CONSOLIDATED_HEADERS, listHeaders);
+			}
 			return getMessageBuilderFactory()
 					.withPayload(payload)
 					.copyHeaders(headers)
@@ -374,16 +392,23 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		@Override
 		public void onMessageBatch(List<Message> messages, Channel channel) {
 			List<?> converted;
+			List<Map<String, Object>> headers = null;
 			if (this.batchModeMessages) {
 				converted = convertMessages(messages, channel);
 			}
 			else {
 				converted = convertPayloads(messages, channel);
+				if (BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS.equals(AmqpInboundChannelAdapter.this.batchMode)) {
+					List<Map<String, Object>> listHeaders = new ArrayList<>();
+					messages.forEach(msg -> listHeaders.add(AmqpInboundChannelAdapter.this.headerMapper
+							.toHeadersFromRequest(msg.getMessageProperties())));
+					headers = listHeaders;
+				}
 			}
 			if (converted != null) {
 				org.springframework.messaging.Message<?> message =
 						createMessageFromPayload(converted, channel, new HashMap<>(),
-								messages.get(messages.size() - 1).getMessageProperties().getDeliveryTag());
+								messages.get(messages.size() - 1).getMessageProperties().getDeliveryTag(), headers);
 				try {
 					if (this.retryOps == null) {
 						setAttributesIfNecessary(messages, message);

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

@@ -449,7 +449,7 @@ public class InboundEndpointTests {
 		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
 		QueueChannel out = new QueueChannel();
 		adapter.setOutputChannel(out);
-		adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS);
+		adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS);
 		adapter.afterPropertiesSet();
 		ChannelAwareBatchMessageListener listener = (ChannelAwareBatchMessageListener) container.getMessageListener();
 		MessageProperties messageProperties = new MessageProperties();
@@ -461,6 +461,8 @@ public class InboundEndpointTests {
 		Message<?> received = out.receive(0);
 		assertThat(received).isNotNull();
 		assertThat(((List<String>) received.getPayload())).contains("test1", "test2");
+		assertThat(received.getHeaders().get(AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS, List.class))
+			.hasSize(2);
 	}
 
 	@SuppressWarnings({ "unchecked" })

src/reference/asciidoc/amqp.adoc

@@ -171,6 +171,7 @@ See the https://docs.spring.io/spring-amqp/reference/html/[Spring AMQP Reference
 <27> When the container's `consumerBatchEnabled` is `true`, determines how the adapter presents the batch of messages in the message payload.
 When set to `MESSAGES` (default), the payload is a `List<Message<?>>` where each message has headers mapped from the incoming AMQP `Message` and the payload is the converted `body`.
 When set to `EXTRACT_PAYLOADS`, the payload is a `List<?>` where the elements are converted from the AMQP `Message` body.
+`EXTRACT_PAYLOADS_WITH_HEADERS` is similar to `EXTRACT_PAYLOADS` but, in addition, the headers from each message are mapped from the `MessageProperties` into a `List<Map<String, Object>` at the corresponding index; the header name is `AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS`.
 ====
 
 [NOTE]

src/reference/asciidoc/whats-new.adoc

@@ -65,3 +65,6 @@ See <<./rmi.adoc#rmi, RMI Support>> for more information.
 
 The outbound endpoints now have a new mechanism for handling publisher confirms and returns.
 See <<./amqp.adoc#alternative-confirms-returns,Alternative Mechanism for Publisher Confirms and Returns>> for more information.
+
+A new `BatchMode.EXTRACT_PAYLOAD_WITH_HEADERS` is supported by the `AmqpInboundChannelAdapter`.
+See <<./amqp.adoc#amqp-inbound-channel-adapter,Inbound Channel Adapter>> for more information.