From 997b07afed231c73c0d056fc6d0b357f396aef54 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 31 May 2018 16:45:07 -0400 Subject: [PATCH] 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. --- .../amqp/support/DefaultAmqpHeaderMapper.java | 27 ++++++++++++++++++- .../support/DefaultAmqpHeaderMapperTests.java | 14 +++++++++- .../mapping/AbstractHeaderMapper.java | 19 +++++++++++-- src/reference/asciidoc/amqp.adoc | 4 +++ src/reference/asciidoc/dsl.adoc | 5 ++-- src/reference/asciidoc/whats-new.adoc | 5 ++++ 6 files changed, 68 insertions(+), 6 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java index e5f76c7a59..35840c20b6 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java +++ b/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,6 +21,7 @@ import java.util.Date; 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; @@ -28,6 +29,8 @@ 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 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper 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 allHeaders, Map headers, + MessageProperties amqpMessageProperties) { String appId = getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class); if (StringUtils.hasText(appId)) { amqpMessageProperties.setAppId(appId); @@ -265,6 +278,12 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper 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(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java index 149e300fcb..e4291491d4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java +++ b/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 java.util.Map.Entry; 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 @@ public abstract class AbstractHeaderMapper implements RequestReplyHeaderMappe 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 @@ public abstract class AbstractHeaderMapper implements RequestReplyHeaderMappe */ protected abstract void populateStandardHeaders(Map 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 allHeaders, Map 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. diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index f8e80d1ccf..6da0e9a935 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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 diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index a30204b220..58411e5e40 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/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 <> for more information. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 41543aa10f..a733a02d1e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -50,3 +50,8 @@ See <> 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 <> for more information. + +==== AMQP Changes + +`ID` and `Timestamp` header mapping changes in the `DefaultAmqpHeaderMapper`. +See the note near the bottom of <> for more information.