Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH- 3223: Build ResolvableType after mapping

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3223

The `json__TypeId__` header may have a value which cannot be resolved to the `Class` in the current classpath
So, skip `ResolvableType` building logic until we really sure that end-user wants to map JSON headers
* WARN a build exception that we can't load a class for the `json__TypeId__` when we try to build a `ResolvableType`
in the  `DefaultAmqpHeaderMapper`
* Document the negation feature for JSON headers

**Cherry-pick to 5.2.x**
This commit is contained in:
Artem Bilan
2020-03-18 17:31:00 -04:00
committed by Gary Russell
parent eea29e729a
commit 3eedda6c9d
4 changed files with 82 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -151,26 +151,13 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
headers.put(jsonHeader, value);
}
}
createJsonResolvableTypHeaderInAny(headers);
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred while mapping from AMQP properties to MessageHeaders", e);
}
this.logger.warn("error occurred while mapping from AMQP properties to MessageHeaders", e);
}
return headers;
}
private void createJsonResolvableTypHeaderInAny(Map<String, Object> headers) {
Object typeIdHeader = headers.get(JsonHeaders.TYPE_ID);
if (typeIdHeader != null) {
headers.put(JsonHeaders.RESOLVABLE_TYPE,
JsonHeaders.buildResolvableType(getClassLoader(), typeIdHeader,
headers.get(JsonHeaders.CONTENT_TYPE_ID), headers.get(JsonHeaders.KEY_TYPE_ID)));
}
}
/**
* Extract user-defined headers from an AMQP MessageProperties instance.
*/

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -26,12 +26,13 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.core.ResolvableType;
@@ -68,7 +69,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void fromHeaders() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
Map<String, Object> headerMap = new HashMap<>();
headerMap.put(AmqpHeaders.APP_ID, "test.appId");
headerMap.put(AmqpHeaders.CLUSTER_ID, "test.clusterId");
headerMap.put(AmqpHeaders.CONTENT_ENCODING, "test.contentEncoding");
@@ -130,7 +131,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void fromHeadersWithContentTypeAsMediaType() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
Map<String, Object> headerMap = new HashMap<>();
MediaType contentType = MediaType.parseMediaType("text/html");
headerMap.put(AmqpHeaders.CONTENT_TYPE, contentType);
@@ -152,7 +153,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void fromHeadersWithContentTypeAsMimeType() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
Map<String, Object> headerMap = new HashMap<>();
MimeType contentType = MimeType.valueOf("text/html");
headerMap.put(AmqpHeaders.CONTENT_TYPE, contentType);
@@ -217,7 +218,7 @@ public class DefaultAmqpHeaderMapperTests {
}
@Test
public void toHeadersNonContenType() {
public void toHeadersNonContentType() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
@@ -244,7 +245,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void messageIdNotMappedToAmqpProperties() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
Map<String, Object> headerMap = new HashMap<>();
headerMap.put(MessageHeaders.ID, "msg-id");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
@@ -255,7 +256,7 @@ public class DefaultAmqpHeaderMapperTests {
@Test
public void messageTimestampNotMappedToAmqpProperties() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
Map<String, Object> headerMap = new HashMap<>();
headerMap.put(MessageHeaders.TIMESTAMP, 1234L);
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
@@ -263,13 +264,13 @@ public class DefaultAmqpHeaderMapperTests {
assertThat(amqpProperties.getHeaders().get(MessageHeaders.TIMESTAMP)).isNull();
}
@Test // INT-2090
@Test
public void jsonTypeIdNotOverwritten() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
MessageConverter converter = new Jackson2JsonMessageConverter();
MessageProperties amqpProperties = new MessageProperties();
converter.toMessage("123", amqpProperties);
Map<String, Object> headerMap = new HashMap<String, Object>();
Map<String, Object> headerMap = new HashMap<>();
headerMap.put("__TypeId__", "java.lang.Integer");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
@@ -325,4 +326,16 @@ public class DefaultAmqpHeaderMapperTests {
assertThat(amqpProperties.getHeaders()).doesNotContainKeys(JsonHeaders.RESOLVABLE_TYPE);
}
@Test
public void jsonHeadersNotMapped() {
DefaultAmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.inboundMapper();
headerMapper.setRequestHeaderNames("!json_*", "*");
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.getHeaders().put(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, "test.type");
Map<String, Object> headers = headerMapper.toHeadersFromRequest(amqpProperties);
assertThat(headers)
.doesNotContainKeys(JsonHeaders.RESOLVABLE_TYPE, JsonHeaders.TYPE_ID)
.containsKey(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME);
}
}

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

@@ -29,6 +29,8 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.core.ResolvableType;
import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
@@ -67,8 +69,8 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
*/
public static final String NON_STANDARD_HEADER_NAME_PATTERN = "NON_STANDARD_HEADERS";
private static final Collection<String> TRANSIENT_HEADER_NAMES = Arrays.asList(
MessageHeaders.ID, MessageHeaders.TIMESTAMP);
private static final Collection<String> TRANSIENT_HEADER_NAMES =
Arrays.asList(MessageHeaders.ID, MessageHeaders.TIMESTAMP);
protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR final
@@ -155,7 +157,7 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
* @return a header mapper that match if any of the specified patters match
*/
protected HeaderMatcher createHeaderMatcher(Collection<String> patterns) {
List<HeaderMatcher> matchers = new ArrayList<HeaderMatcher>();
List<HeaderMatcher> matchers = new ArrayList<>();
for (String pattern : patterns) {
if (STANDARD_REQUEST_HEADER_NAME_PATTERN.equals(pattern)) {
matchers.add(new ContentBasedHeaderMatcher(true, this.requestHeaderNames));
@@ -190,40 +192,38 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
@Override
public void fromHeadersToRequest(MessageHeaders headers, T target) {
this.fromHeaders(headers, target, this.requestHeaderMatcher);
fromHeaders(headers, target, this.requestHeaderMatcher);
}
@Override
public void fromHeadersToReply(MessageHeaders headers, T target) {
this.fromHeaders(headers, target, this.replyHeaderMatcher);
fromHeaders(headers, target, this.replyHeaderMatcher);
}
@Override
public Map<String, Object> toHeadersFromRequest(T source) {
return this.toHeaders(source, this.requestHeaderMatcher);
return toHeaders(source, this.requestHeaderMatcher);
}
@Override
public Map<String, Object> toHeadersFromReply(T source) {
return this.toHeaders(source, this.replyHeaderMatcher);
return toHeaders(source, this.replyHeaderMatcher);
}
private void fromHeaders(MessageHeaders headers, T target, HeaderMatcher headerMatcher) {
try {
Map<String, Object> subset = new HashMap<String, Object>();
Map<String, Object> subset = new HashMap<>();
for (Map.Entry<String, Object> entry : headers.entrySet()) {
String headerName = entry.getKey();
if (this.shouldMapHeader(headerName, headerMatcher)) {
if (shouldMapHeader(headerName, headerMatcher)) {
subset.put(headerName, entry.getValue());
}
}
this.populateStandardHeaders(headers, subset, target);
this.populateUserDefinedHeaders(subset, target);
populateStandardHeaders(headers, subset, target);
populateUserDefinedHeaders(subset, target);
}
catch (Exception e) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("error occurred while mapping from MessageHeaders", e);
}
this.logger.warn("error occurred while mapping from MessageHeaders", e);
}
}
@@ -234,8 +234,8 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
if (value != null && !isMessageChannel(headerName, value)) {
try {
if (!headerName.startsWith(this.standardHeaderPrefix)) {
String key = this.createTargetPropertyName(headerName, true);
this.populateUserDefinedHeader(key, value, target);
String key = createTargetPropertyName(headerName, true);
populateUserDefinedHeader(key, value, target);
}
}
catch (Exception e) {
@@ -262,21 +262,30 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
* a {@link org.springframework.messaging.Message}.
*/
private Map<String, Object> toHeaders(T source, HeaderMatcher headerMatcher) {
Map<String, Object> headers = new HashMap<String, Object>();
Map<String, Object> headers = new HashMap<>();
Map<String, Object> standardHeaders = extractStandardHeaders(source);
this.copyHeaders(standardHeaders, headers, headerMatcher);
copyHeaders(standardHeaders, headers, headerMatcher);
Map<String, Object> userDefinedHeaders = extractUserDefinedHeaders(source);
this.copyHeaders(userDefinedHeaders, headers, headerMatcher);
copyHeaders(userDefinedHeaders, headers, headerMatcher);
return headers;
}
private <V> void copyHeaders(Map<String, Object> source, Map<String, Object> target, HeaderMatcher headerMatcher) {
private void copyHeaders(Map<String, Object> source, Map<String, Object> target, HeaderMatcher headerMatcher) {
if (!CollectionUtils.isEmpty(source)) {
for (Map.Entry<String, Object> entry : source.entrySet()) {
try {
String headerName = this.createTargetPropertyName(entry.getKey(), false);
if (this.shouldMapHeader(headerName, headerMatcher)) {
target.put(headerName, entry.getValue());
String headerName = createTargetPropertyName(entry.getKey(), false);
if (shouldMapHeader(headerName, headerMatcher)) {
Object value = entry.getValue();
target.put(headerName, value);
if (JsonHeaders.TYPE_ID.equals(headerName) && value != null) {
ResolvableType resolvableType =
createJsonResolvableTypHeaderInAny(value, source.get(JsonHeaders.CONTENT_TYPE_ID),
source.get(JsonHeaders.KEY_TYPE_ID));
if (resolvableType != null) {
target.put(JsonHeaders.RESOLVABLE_TYPE, resolvableType);
}
}
}
}
catch (Exception e) {
@@ -289,6 +298,20 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
}
}
@Nullable
private ResolvableType createJsonResolvableTypHeaderInAny(Object typeId, @Nullable Object contentId,
@Nullable Object keyId) {
try {
return JsonHeaders.buildResolvableType(getClassLoader(), typeId,
contentId, keyId);
}
catch (Exception e) {
this.logger.warn("Cannot build a ResolvableType from 'json__TypeId__' header", e);
}
return null;
}
private boolean shouldMapHeader(String headerName, HeaderMatcher headerMatcher) {
return !(!StringUtils.hasText(headerName) || getTransientHeaderNames().contains(headerName))
&& headerMatcher.matchHeader(headerName);
@@ -303,8 +326,8 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
}
if (!type.isAssignableFrom(value.getClass())) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "], it is [" +
value.getClass() + "]");
this.logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "], " +
"it is [" + value.getClass() + "]");
}
return null;
}
@@ -380,6 +403,7 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
* Strategy interface to determine if a given header name matches.
* @since 4.1
*/
@FunctionalInterface
public interface HeaderMatcher {
/**
@@ -393,7 +417,9 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
* Return true if this match should be explicitly excluded from the mapping.
* @return true if negated.
*/
boolean isNegated();
default boolean isNegated() {
return false;
}
}
@@ -440,11 +466,6 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
return false;
}
@Override
public boolean isNegated() {
return false;
}
}
/**
@@ -457,7 +478,7 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
private static final Log logger = LogFactory.getLog(HeaderMatcher.class);
private final Collection<String> patterns = new ArrayList<String>();
private final Collection<String> patterns = new ArrayList<>();
public PatternBasedHeaderMatcher(Collection<String> patterns) {
Assert.notNull(patterns, "Patterns must no be null");
@@ -482,11 +503,6 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
return false;
}
@Override
public boolean isNegated() {
return false;
}
}
/**
@@ -566,11 +582,6 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
return result;
}
@Override
public boolean isNegated() {
return false;
}
}
/**
@@ -608,11 +619,6 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
return false;
}
@Override
public boolean isNegated() {
return false;
}
}
}

10
src/reference/asciidoc/amqp.adoc
View File

@@ -1436,10 +1436,8 @@ The `org.springframework.amqp.support.AmqpHeaders` class identifies the default
[[header-copy-caution]]
CAUTION: As mentioned earlier in this section, using a header mapping pattern of`*` is a common way to copy all headers.
However, this can have some unexpected side effects, because certain RabbitMQ proprietary properties/headers are also
copied.
For example, when you use https://www.rabbitmq.com/federated-exchanges.html[federation], the received message may have
a property named `x-received-from`, which contains the node that sent the message.
However, this can have some unexpected side effects, because certain RabbitMQ proprietary properties/headers are also copied.
For example, when you use https://www.rabbitmq.com/federated-exchanges.html[federation], the received message may have a property named `x-received-from`, which contains the node that sent the message.
If you use the wildcard character `*` for the request and reply header mapping on the inbound gateway, this header is copied, which may cause some issues with federation.
This reply message may be federated back to the sending broker, which may think that a message is looping and, as a result, silently drop it.
If you wish to use the convenience of wildcard header mapping, you may need to filter out some headers in the downstream flow.
@@ -1452,6 +1450,8 @@ Instead, this header is mapped to `amqp_receivedDeliveryMode`, which is not mapp
Starting with version 4.3, patterns in the header mappings can be negated by preceding the pattern with `!`.
Negated patterns get priority, so a list such as `STANDARD_REQUEST_HEADERS,thing1,ba*,!thing2,!thing3,qux,!thing1` does not map `thing1` (nor `thing2` nor `thing3`).
The standard headers plus `bad` and `qux` are mapped.
The negation technique can be useful for example to not map JSON type headers for incoming messages when a JSON deserialization logic is done in the receiver downstream different way.
For this purpose a `!json_*` pattern should be configured for header mapper of the inbound channel adapter/gateway.
IMPORTANT: If you have a user-defined header that begins with `!` that you do wish to map, you need to escape it with `\`, as follows: `STANDARD_REQUEST_HEADERS,\!myBangHeader`.
The header named `!myBangHeader` is now mapped.
@@ -1468,7 +1468,7 @@ For example an inbound HTTP message sent to a RabbitMQ queue.
The `contentType` header is mapped to Spring AMQP's `MessageProperties.contentType` property and that is subsequently mapped to RabbitMQ's `content_type` property.
Prior to _version 5.1_, this header was also mapped as an entry in the `MessageProperties.headers` map; this was incorrect and, furthermore, the value could be wrong since the underlying Spring AMQP message converter might have changed the content type.
Prior to version 5.1, this header was also mapped as an entry in the `MessageProperties.headers` map; this was incorrect and, furthermore, the value could be wrong since the underlying Spring AMQP message converter might have changed the content type.
Such a change would be reflected in the first-class `content_type` property, but not in the RabbitMQ headers map.
Inbound mapping ignored the headers map value.
`contentType` is no longer mapped to an entry in the headers map.