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

DefaultAmqpHeaderMapper refactoring

Browse Source 
INT-2016, INT-2017 transient headers are no longer mapped
  INT-2090 prevent overwriting of existing headers

Most of INT-2016 is already covered by the other changes made for INT-2017.
However, it was mentioned in INT-2016 that the JsonMessageConverter was
setting a __TypeId__ header, and that header was then being overwritten
if one was present in the MessageHeaders. This change now prevents any
header value from being overwritten so that the MessageConverter or
any process that explicitly added headers to the Message would take
precedence over the HeaderMapper which may write headers subsequently.

excluding id, timestamp, error/reply channel only

updated comment issue link
This commit is contained in:
Mark Fisher
2011-08-25 14:24:41 -04:00
parent 974b385a08
commit a832796d00
2 changed files with 106 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,11 +23,13 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
@@ -43,9 +45,18 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
private static final String[] TRANSIENT_HEADER_NAMES = new String[] {
MessageHeaders.ID,
MessageHeaders.ERROR_CHANNEL,
MessageHeaders.REPLY_CHANNEL,
MessageHeaders.TIMESTAMP
};
private final Log logger = LogFactory.getLog(this.getClass());
private volatile String inboundPrefix = "";
@@ -164,22 +175,24 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
// now map to the user-defined headers, if any, within the AMQP MessageProperties
Set<String> headerNames = headers.keySet();
for (String headerName : headerNames) {
if (this.shouldMapHeader(headerName)){
if (StringUtils.hasText(headerName) && !headerName.startsWith(AmqpHeaders.PREFIX)) {
Object value = headers.get(headerName);
if (value != null) {
try {
String key = this.fromHeaderName(headerName);
if (this.shouldMapOutboundHeader(headerName)) {
Object value = headers.get(headerName);
if (value != null) {
try {
String key = this.fromHeaderName(headerName);
// do not overwrite an existing header with the same key
// TODO: do we need to expose a boolean 'overwrite' flag?
if (!amqpMessageProperties.getHeaders().containsKey(key)) {
amqpMessageProperties.setHeader(key, value);
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to map Message header '" + headerName + "' to AMQP header", e);
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to map Message header '" + headerName + "' to AMQP header", e);
}
}
}
}
}
}
}
catch (Exception e) {
@@ -188,13 +201,6 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
}
}
}
private boolean shouldMapHeader(String headerName){
return !headerName.equals(MessageHeaders.ERROR_CHANNEL) &&
!headerName.equals(MessageHeaders.REPLY_CHANNEL) &&
!headerName.equals(MessageHeaders.ID) &&
!headerName.equals(MessageHeaders.TIMESTAMP);
}
/**
* Maps headers from an AMQP MessageProperties instance to the MessageHeaders of a
@@ -284,7 +290,9 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
for (Map.Entry<String, Object> entry : amqpHeaders.entrySet()) {
try {
String headerName = this.toHeaderName(entry.getKey());
headers.put(headerName, entry.getValue());
if (!ObjectUtils.containsElement(TRANSIENT_HEADER_NAMES, headerName)) {
headers.put(headerName, entry.getValue());
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
@@ -303,6 +311,12 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
return headers;
}
private boolean shouldMapOutboundHeader(String headerName) {
return StringUtils.hasText(headerName)
&& !headerName.startsWith(AmqpHeaders.PREFIX)
&& !ObjectUtils.containsElement(TRANSIENT_HEADER_NAMES, headerName);
}
private <T> T getHeaderIfAvailable(MessageHeaders headers, String name, Class<T> type) {
try {
return headers.get(name, type);

73
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java Normal file
View File

@@ -0,0 +1,73 @@
/*
* Copyright 2002-2011 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.amqp.support;
import static org.junit.Assert.assertEquals;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.JsonMessageConverter;
import org.springframework.integration.MessageHeaders;
/**
* @author Mark Fisher
* @since 2.1
*/
public class DefaultAmqpHeaderMapperTests {
@Test
public void replyChannelNotMappedToAmqpProperties() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put(MessageHeaders.REPLY_CHANNEL, "foo");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
headerMapper.fromHeaders(integrationHeaders, amqpProperties);
assertEquals(null, amqpProperties.getHeaders().get(MessageHeaders.REPLY_CHANNEL));
}
@Test
public void errorChannelNotMappedToAmqpProperties() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put(MessageHeaders.ERROR_CHANNEL, "foo");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
headerMapper.fromHeaders(integrationHeaders, amqpProperties);
assertEquals(null, amqpProperties.getHeaders().get(MessageHeaders.ERROR_CHANNEL));
}
@Test // INT-2090
public void jsonTypeIdNotOverwritten() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
JsonMessageConverter converter = new JsonMessageConverter();
MessageProperties amqpProperties = new MessageProperties();
converter.toMessage("123", amqpProperties);
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put("__TypeId__", "java.lang.Integer");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
headerMapper.fromHeaders(integrationHeaders, amqpProperties);
assertEquals("java.lang.String", amqpProperties.getHeaders().get("__TypeId__"));
Object result = converter.fromMessage(new Message("123".getBytes(), amqpProperties));
assertEquals(String.class, result.getClass());
}
}