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 a848d6840d..509fae94db 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 @@ -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 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 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 getHeaderIfAvailable(MessageHeaders headers, String name, Class type) { try { return headers.get(name, type); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java new file mode 100644 index 0000000000..3943e46e5d --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java @@ -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 headerMap = new HashMap(); + 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 headerMap = new HashMap(); + 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 headerMap = new HashMap(); + 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()); + } +}