INT-4476: Fall back for ID and Timestamp Headers

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

When mapping outbound headers, if the `AmqpHeaders` `ID` and `Timestamp` headers
are not present, fall back to mapping the `MessageHeaders` variants (if present).

Also fix some PDF overflows.

Author: garyrussell

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,13 +21,16 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.mapping.AbstractHeaderMapper;
 import org.springframework.integration.mapping.support.JsonHeaders;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.MimeType;
 import org.springframework.util.StringUtils;
 
@@ -213,6 +216,16 @@ protected Map<String, Object> extractUserDefinedHeaders(MessageProperties amqpMe
 	 */
 	@Override
 	protected void populateStandardHeaders(Map<String, Object> headers, MessageProperties amqpMessageProperties) {
+		populateStandardHeaders(null, headers, amqpMessageProperties);
+	}
+
+	/**
+	 * Maps headers from a Spring Integration MessageHeaders instance to the MessageProperties
+	 * of an AMQP Message.
+	 */
+	@Override
+	protected void populateStandardHeaders(@Nullable Map<String, Object> allHeaders, Map<String, Object> headers,
+			MessageProperties amqpMessageProperties) {
 		String appId = getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class);
 		if (StringUtils.hasText(appId)) {
 			amqpMessageProperties.setAppId(appId);
@@ -265,6 +278,12 @@ protected void populateStandardHeaders(Map<String, Object> headers, MessagePrope
 		if (StringUtils.hasText(messageId)) {
 			amqpMessageProperties.setMessageId(messageId);
 		}
+		else if (allHeaders != null) {
+			UUID id = getHeaderIfAvailable(allHeaders, MessageHeaders.ID, UUID.class);
+			if (id != null) {
+				amqpMessageProperties.setMessageId(id.toString());
+			}
+		}
 		Integer priority = getHeaderIfAvailable(headers, IntegrationMessageHeaderAccessor.PRIORITY, Integer.class);
 		if (priority != null) {
 			amqpMessageProperties.setPriority(priority);
@@ -289,6 +308,12 @@ protected void populateStandardHeaders(Map<String, Object> headers, MessagePrope
 		if (timestamp != null) {
 			amqpMessageProperties.setTimestamp(timestamp);
 		}
+		else if (allHeaders != null) {
+			Long ts = getHeaderIfAvailable(allHeaders, MessageHeaders.TIMESTAMP, Long.class);
+			if (ts != null) {
+				amqpMessageProperties.setTimestamp(new Date(ts));
+			}
+		}
 		String type = getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class);
 		if (type != null) {
 			amqpMessageProperties.setType(type);

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.amqp.support;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.fail;
@@ -37,6 +38,7 @@
 import org.springframework.http.MediaType;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
@@ -50,6 +52,16 @@
  */
 public class DefaultAmqpHeaderMapperTests {
 
+	@Test
+	public void fromHeadersFallbackIdTimestamp() {
+		DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
+		org.springframework.messaging.Message<?> message = new GenericMessage<>("");
+		MessageProperties messageProperties = new MessageProperties();
+		headerMapper.fromHeadersToRequest(message.getHeaders(), messageProperties);
+		assertThat(message.getHeaders().getId().toString()).isEqualTo(messageProperties.getMessageId());
+		assertThat(message.getHeaders().getTimestamp()).isEqualTo(messageProperties.getTimestamp().getTime());
+	}
+
 	@Test
 	public void fromHeaders() {
 		DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();

spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
@@ -201,7 +202,7 @@ private void fromHeaders(MessageHeaders headers, T target, HeaderMatcher headerM
 					subset.put(headerName, entry.getValue());
 				}
 			}
-			this.populateStandardHeaders(subset, target);
+			this.populateStandardHeaders(headers, subset, target);
 			this.populateUserDefinedHeaders(subset, target);
 		}
 		catch (Exception e) {
@@ -337,6 +338,20 @@ protected Collection<String> getTransientHeaderNames() {
 	 */
 	protected abstract void populateStandardHeaders(Map<String, Object> headers, T target);
 
+	/**
+	 * Populate the specified standard headers to the specified source.
+	 * If not implemented, calls {@link #populateStandardHeaders(Map, Object)}.
+	 * @param allHeaders all headers including transient.
+	 * @param subset the map of standard headers to be populated.
+	 * @param target the target object to populate headers.
+	 * @since 5.1
+	 */
+	protected void populateStandardHeaders(@Nullable Map<String, Object> allHeaders, Map<String, Object> subset,
+			T target) {
+
+		populateStandardHeaders(subset, target);
+	}
+
 	/**
 	 * Populate the specified user-defined headers to the specified source.
 	 * @param headerName the user defined header name to be populated.

src/reference/asciidoc/amqp.adoc

@@ -1504,6 +1504,10 @@ Negated patterns get priority, so a list such as
 IMPORTANT: If you have a user defined header that begins with `!` that you *do* wish to map, you need to escape it with
 `\` thus: `STANDARD_REQUEST_HEADERS,\!myBangHeader` and it *WILL* be mapped.
 
+NOTE: Starting with _version 5.1_, the `DefaultAmqpHeaderMapper` will fall back to mapping `MessageHeaders.ID` and `MessageHeaders.TIMESTAMP` to `MessageProperties.messageId` and `MessageProperties.timestamp` respectively, if the corresponding `amqp_messageId` or `amqp_timestamp` headers are not present on outbound messages.
+Inbound properties will be mapped to the `amqp_*` headers as before.
+It is useful to populate the `messageId` property when message consumers are using stateful retry.
+
 [[amqp-strict-ordering]]
 === Strict Message Ordering
 

src/reference/asciidoc/dsl.adoc

@@ -581,8 +581,9 @@ And Lambda flow can't start from `MessageSource` or `MessageProducer`.
 Starting _version 5.1_, this kind of `IntegrationFlow` are wrapped to the proxy for exposing lifecycle control and provide access to the `inputChannel` of the internally associated `StandardIntegrationFlow`.
 
 Starting with _version 5.0.6_, the generated bean names for the components in an `IntegrationFlow` include the flow bean followed by a dot as a prefix.
-For example the `ConsumerEndpointFactoryBean` for the `.transform("Hello "::concat)` in the sample above, will end up with te bean name like `lambdaFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0`.
-The `Transformer` implementation bean for that endpoint will have a bean name such as `lambdaFlow.org.springframework.integration.transformer.MethodInvokingTransformer#0`.
+For example the `ConsumerEndpointFactoryBean` for the `.transform("Hello "::concat)` in the sample above, will result with a bean name `lambdaFlow.o.s.integration.config.ConsumerEndpointFactoryBean#0`.
+The `Transformer` implementation bean for that endpoint will have a bean name `lambdaFlow.o.s.integration.transformer.MethodInvokingTransformer#0`.
+(In both cases, `o.s` is `org.springframework`; shortened here to fit on the page).
 These generated bean names are prepended with the flow id prefix for purposes such as parsing logs or grouping components together in some analysis tool, as well as to avoid a race condition when we concurrently register integration flows at runtime.
 See <<java-dsl-runtime-flows>> for more information.
 

src/reference/asciidoc/whats-new.adoc

@@ -50,3 +50,8 @@ See <<json-transformers>> for more information.
 Starting with _version 5.0.5_, generated bean names for the components in an `IntegrationFlow` include the flow bean name, followed by a dot, as a prefix.
 
 See <<java-dsl-flows>> for more information.
+
+==== AMQP Changes
+
+`ID` and `Timestamp` header mapping changes in the `DefaultAmqpHeaderMapper`.
+See the note near the bottom of <<amqp-message-headers>> for more information.