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

Merge pull request #38 from markfisher/INT-2017

Browse Source 
INT-2016, INT-2017, INT-2090:
  DefaultAmqpHeaderMapper refactoring
This commit is contained in:
Mark Fisher
2011-08-30 15:42:49 -04:00
parent 974b385a08 a832796d00
commit f4cb32c866
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());
}
}