diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index 4b164f3191..d6952fe1a4 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/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} 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 createMessageFromPayload(Object payload, - Channel channel, Map headers, long deliveryTag) { + Channel channel, Map headers, long deliveryTag, + @Nullable List> 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 messages, Channel channel) { List converted; + List> 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> 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); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index d28235902f..de02836854 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/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) received.getPayload())).contains("test1", "test2"); + assertThat(received.getHeaders().get(AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS, List.class)) + .hasSize(2); } @SuppressWarnings({ "unchecked" }) diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 9191e729d8..61acebf530 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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>` 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` at the corresponding index; the header name is `AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS`. ==== [NOTE] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 4cb319a71a..5edbf18713 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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.