From 6dbc721d732c895f5779b5f60a82dcbcc50be51b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 2 Apr 2017 16:12:43 -0400 Subject: [PATCH] INT-4249: AMQP: Configurable Header Mapping Order JIRA: https://jira.spring.io/browse/INT-4249 Add `headersMappedLast` to outbound endpoints and channels. * Fix typos in JavaDocs and `amqp.adoc` --- .../amqp/channel/AbstractAmqpChannel.java | 28 ++++++- .../amqp/config/AmqpChannelFactoryBean.java | 9 +- .../amqp/config/AmqpChannelParser.java | 3 +- .../AmqpOutboundChannelAdapterParser.java | 3 +- .../config/AmqpOutboundGatewayParser.java | 3 +- .../dsl/AmqpBaseOutboundEndpointSpec.java | 10 +++ .../dsl/AmqpPollableMessageChannelSpec.java | 10 +++ .../AbstractAmqpOutboundEndpoint.java | 24 ++++++ .../amqp/outbound/AmqpOutboundEndpoint.java | 4 +- .../outbound/AsyncAmqpOutboundGateway.java | 2 +- .../amqp/support/MappingUtils.java | 41 +++++++-- .../config/spring-integration-amqp-5.0.xsd | 20 ++++- .../config/AmqpChannelParserTests-context.xml | 1 + .../amqp/config/AmqpChannelParserTests.java | 4 +- ...boundChannelAdapterParserTests-context.xml | 1 + ...AmqpOutboundChannelAdapterParserTests.java | 3 + ...AmqpOutboundGatewayParserTests-context.xml | 1 + .../AmqpOutboundGatewayParserTests.java | 10 ++- .../amqp/outbound/OutboundEndpointTests.java | 2 + .../amqp/support/MappingUtilsTests.java | 84 +++++++++++++++++++ src/reference/asciidoc/amqp.adoc | 16 ++++ src/reference/asciidoc/whats-new.adoc | 4 + 22 files changed, 261 insertions(+), 22 deletions(-) create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/MappingUtilsTests.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java index 4605d9cbae..213c991422 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -19,6 +19,7 @@ package org.springframework.integration.amqp.channel; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.AmqpHeaders; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.amqp.support.MappingUtils; @@ -46,6 +47,8 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { private MessageDeliveryMode defaultDeliveryMode; + private boolean headersMappedLast; + /** * Construct an instance with the supplied template and default header mappers * used if the template is a {@link RabbitTemplate} and the message is mapped. @@ -123,6 +126,26 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { return this.extractPayload; } + /** + * When mapping headers for the outbound message, determine whether the headers are + * mapped before the message is converted, or afterwards. This only affects headers + * that might be added by the message converter. When false, the converter's headers + * win; when true, any headers added by the converter will be overridden (if the + * source message has a header that maps to those headers). You might wish to set this + * to true, for example, when using a + * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} with a + * String payload that contains json; the converter will set the content type to + * {@code text/plain} which can be overridden to {@code application/json} by setting + * the {@link AmqpHeaders#CONTENT_TYPE} message header. + * Only applies when {@link #setExtractPayload(boolean) extractPayload} is true. + * Default: false. + * @param headersMappedLast true if headers are mapped after conversion. + * @since 5.0 + */ + public void setHeadersMappedLast(boolean headersMappedLast) { + this.headersMappedLast = headersMappedLast; + } + /** * Subclasses may override this method to return an Exchange name. * By default, Messages will be sent to the no-name Direct Exchange. @@ -159,7 +182,8 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel { protected boolean doSend(Message message, long timeout) { if (this.extractPayload) { this.amqpTemplate.send(getExchangeName(), getRoutingKey(), MappingUtils.mapMessage(message, - this.rabbitTemplate.getMessageConverter(), this.outboundHeaderMapper, this.defaultDeliveryMode)); + this.rabbitTemplate.getMessageConverter(), this.outboundHeaderMapper, this.defaultDeliveryMode, + this.headersMappedLast)); } else { this.amqpTemplate.convertAndSend(getExchangeName(), getRoutingKey(), message); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java index 44956bfaa9..5b86987dbc 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -139,6 +139,8 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean getObjectType() { return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class; @@ -401,6 +407,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean delayGenerator; + private boolean headersMappedLast; + private volatile boolean running; public void setHeaderMapper(AmqpHeaderMapper headerMapper) { @@ -91,6 +93,24 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin this.headerMapper = headerMapper; } + /** + * When mapping headers for the outbound message, determine whether the headers are + * mapped before the message is converted, or afterwards. This only affects headers + * that might be added by the message converter. When false, the converter's headers + * win; when true, any headers added by the converter will be overridden (if the + * source message has a header that maps to those headers). You might wish to set this + * to true, for example, when using a + * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} with a + * String payload that contains json; the converter will set the content type to + * {@code text/plain} which can be overridden to {@code application/json} by setting + * the {@link AmqpHeaders#CONTENT_TYPE} message header. Default: false. + * @param headersMappedLast true if headers are mapped after conversion. + * @since 5.0 + */ + public void setHeadersMappedLast(boolean headersMappedLast) { + this.headersMappedLast = headersMappedLast; + } + public void setExchangeName(String exchangeName) { Assert.notNull(exchangeName, "exchangeName must not be null"); this.exchangeName = exchangeName; @@ -293,6 +313,10 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin return this.lazyConnect; } + protected boolean isHeadersMappedLast() { + return this.headersMappedLast; + } + @Override protected final void doInit() { Assert.state(this.exchangeNameExpression == null || this.exchangeName == null, diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index 696d3997ba..f2e1f73ba6 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -102,7 +102,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint if (this.amqpTemplate instanceof RabbitTemplate) { MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, - getHeaderMapper(), getDefaultDeliveryMode()); + getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast()); addDelayProperty(requestMessage, amqpMessage); ((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData); } @@ -122,7 +122,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint "RabbitTemplate implementation is required for publisher confirms"); MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, - getHeaderMapper(), getDefaultDeliveryMode()); + getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast()); addDelayProperty(requestMessage, amqpMessage); org.springframework.amqp.core.Message amqpReplyMessage = ((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage, diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index b31c2cf305..9d39ec4f33 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -76,7 +76,7 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { @Override protected Object handleRequestMessage(Message requestMessage) { org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, - this.messageConverter, getHeaderMapper(), getDefaultDeliveryMode()); + this.messageConverter, getHeaderMapper(), getDefaultDeliveryMode(), isHeadersMappedLast()); addDelayProperty(requestMessage, amqpMessage); RabbitMessageFuture future = this.template.sendAndReceive(generateExchangeName(requestMessage), generateRoutingKey(requestMessage), amqpMessage); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java index 4d824cf9e9..25fab96761 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -22,6 +22,8 @@ import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.MimeType; /** * Utility methods used during message mapping. @@ -37,29 +39,54 @@ public final class MappingUtils { } /** - * Map an o.s.Message to an o.s.a.core.Message. + * Map an o.s.Message to an o.s.a.core.Message. When using a + * {@link ContentTypeDelegatingMessageConverter}, {@link AmqpHeaders#CONTENT_TYPE} and + * {@link MessageHeaders#CONTENT_TYPE} will be used for the selection, with the AMQP + * header taking precedence. * @param requestMessage the request message. * @param converter the message converter to use. * @param headerMapper the header mapper to use. * @param defaultDeliveryMode the default delivery mode. + * @param headersMappedLast true if headers are mapped after conversion. * @return the mapped Message. */ public static org.springframework.amqp.core.Message mapMessage(Message requestMessage, - MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode) { + MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode, + boolean headersMappedLast) { MessageProperties amqpMessageProperties = new MessageProperties(); org.springframework.amqp.core.Message amqpMessage; - if (converter instanceof ContentTypeDelegatingMessageConverter) { + if (!headersMappedLast) { headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); - amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); } - else { // See INT-3002 - map headers last if we're not using a CTDMC - amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); + if (converter instanceof ContentTypeDelegatingMessageConverter && headersMappedLast) { + String contentType = contentTypeAsString(requestMessage.getHeaders()); + if (contentType != null) { + amqpMessageProperties.setContentType(contentType); + } + } + amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); + if (headersMappedLast) { headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); } checkDeliveryMode(requestMessage, amqpMessageProperties, defaultDeliveryMode); return amqpMessage; } + private static String contentTypeAsString(MessageHeaders headers) { + Object contentType = headers.get(AmqpHeaders.CONTENT_TYPE); + if (contentType instanceof MimeType) { + contentType = contentType.toString(); + } + if (contentType instanceof String) { + return (String) contentType; + } + else if (contentType != null) { + throw new IllegalArgumentException(AmqpHeaders.CONTENT_TYPE + + " header must be a MimeType or String, found: " + contentType.getClass().getName()); + } + return null; + } + /** * Check the delivery mode and update with the default if not already present. * @param requestMessage the request message. diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.0.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.0.xsd index 00feaadd81..33d98eab40 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.0.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.0.xsd @@ -364,6 +364,7 @@ Set to 'true' to extract the message payload and map the o.s.messaging.Message to an o.s.amqp.core.Message in a similar manner to a pair of channel adapters. When 'false' the entire message is converted requiring either Java serializable contents or a custom message converter. Also see inbound and outbound mapped headers. + Also see 'headers-last', which only applies if this is 'true'. @@ -384,7 +385,8 @@ - + + @@ -557,6 +559,7 @@ property set to TRUE. + @@ -1014,6 +1017,21 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA + + + + + Whether headers are mapped before or after conversion from a messaging Message to + a spring amqp Message. Set to true, for example, if you wish to override the + contentType header set by the converter. + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml index 2633d4b12f..58b9654ce4 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml @@ -26,6 +26,7 @@ inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" /> diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java index ab37107fcb..564e2c98fe 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java @@ -128,6 +128,7 @@ public class AmqpOutboundChannelAdapterParserTests { assertEquals("42", TestUtils.getPropertyValue(endpoint, "delayExpression", org.springframework.expression.Expression.class) .getExpressionString()); + assertFalse(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)); Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate"); amqpTemplateField.setAccessible(true); @@ -297,6 +298,8 @@ public class AmqpOutboundChannelAdapterParserTests { AmqpHeaderMapper headerMapper = TestUtils.getPropertyValue(this.amqpMessageHandlerWithCustomHeaderMapper, "headerMapper", AmqpHeaderMapper.class); assertSame(this.context.getBean("customHeaderMapper"), headerMapper); + assertTrue(TestUtils.getPropertyValue(this.amqpMessageHandlerWithCustomHeaderMapper, + "headersMappedLast", Boolean.class)); } @Test diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml index f7c1ef8b0b..64903d9aed 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml @@ -49,6 +49,7 @@ order="5" default-delivery-mode="NON_PERSISTENT" requires-reply="false" + headers-last="true" mapped-request-headers="foo*" mapped-reply-headers="bar*"/> diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index 5607512b1b..a2942da215 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -101,14 +101,15 @@ public class AmqpOutboundGatewayParserTests { public void withHeaderMapperCustomRequestResponse() { ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomRequestResponse"); + Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomRequestResponse"); - AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", + AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class); assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode")); assertFalse(TestUtils.getPropertyValue(endpoint, "lazyConnect", Boolean.class)); assertFalse(TestUtils.getPropertyValue(endpoint, "requiresReply", Boolean.class)); + assertTrue(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)); Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate"); amqpTemplateField.setAccessible(true); @@ -170,11 +171,12 @@ public class AmqpOutboundGatewayParserTests { public void withHeaderMapperCustomAndStandardResponse() { ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse"); + Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse"); - AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", + AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class); assertNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode")); + assertFalse(TestUtils.getPropertyValue(endpoint, "headersMappedLast", Boolean.class)); Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate"); amqpTemplateField.setAccessible(true); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java index abb39a5045..7293269d0e 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java @@ -119,6 +119,7 @@ public class OutboundEndpointTests { ConnectionFactory connectionFactory = mock(ConnectionFactory.class); RabbitTemplate amqpTemplate = spy(new RabbitTemplate(connectionFactory)); AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate); + endpoint.setHeadersMappedLast(true); final AtomicReference amqpMessage = new AtomicReference(); willAnswer(invocation -> { @@ -138,6 +139,7 @@ public class OutboundEndpointTests { ConnectionFactory connectionFactory = mock(ConnectionFactory.class); TestRabbitTemplate amqpTemplate = spy(new TestRabbitTemplate(connectionFactory)); AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate); + endpoint.setHeadersMappedLast(true); endpoint.setExpectReply(true); DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); mapper.setRequestHeaderNames("*"); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/MappingUtilsTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/MappingUtilsTests.java new file mode 100644 index 0000000000..2e9eee39e2 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/MappingUtilsTests.java @@ -0,0 +1,84 @@ +/* + * Copyright 2017 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.hamcrest.CoreMatchers.equalTo; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.fail; + +import org.junit.Test; + +import org.springframework.amqp.core.MessageDeliveryMode; +import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +/** + * @author Gary Russell + * @since 5.0 + * + */ +public class MappingUtilsTests { + + @Test + public void testMapping() { + Message requestMessage = MessageBuilder.withPayload("foo") + .setHeader(AmqpHeaders.CONTENT_TYPE, "my/ct") + .build(); + MessageConverter converter = new SimpleMessageConverter(); + AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); + MessageDeliveryMode defaultDeliveryMode = MessageDeliveryMode.NON_PERSISTENT; + boolean headersMappedLast = false; + org.springframework.amqp.core.Message mapped = MappingUtils.mapMessage(requestMessage, converter, headerMapper, + defaultDeliveryMode, headersMappedLast); + assertThat(mapped.getMessageProperties().getContentType(), equalTo("text/plain")); + + headersMappedLast = true; + mapped = MappingUtils.mapMessage(requestMessage, converter, headerMapper, + defaultDeliveryMode, headersMappedLast); + assertThat(mapped.getMessageProperties().getContentType(), equalTo("my/ct")); + + ContentTypeDelegatingMessageConverter ctdConverter = new ContentTypeDelegatingMessageConverter(); + ctdConverter.addDelegate("my/ct", converter); + mapped = MappingUtils.mapMessage(requestMessage, ctdConverter, headerMapper, + defaultDeliveryMode, headersMappedLast); + assertThat(mapped.getMessageProperties().getContentType(), equalTo("my/ct")); + + headersMappedLast = false; + mapped = MappingUtils.mapMessage(requestMessage, ctdConverter, headerMapper, + defaultDeliveryMode, headersMappedLast); + assertThat(mapped.getMessageProperties().getContentType(), equalTo("text/plain")); + + headersMappedLast = true; + requestMessage = MessageBuilder.withPayload("foo") + .setHeader(AmqpHeaders.CONTENT_TYPE, 42) + .build(); + try { + mapped = MappingUtils.mapMessage(requestMessage, ctdConverter, headerMapper, + defaultDeliveryMode, headersMappedLast); + fail("Expected IllegalArgumentException"); + } + catch (IllegalArgumentException e) { + assertThat(e.getMessage(), + equalTo("contentType header must be a MimeType or String, found: java.lang.Integer")); + } + } + +} diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 7f2fe5608d..6ec9d2e03c 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1140,6 +1140,22 @@ JSON converter to be selected. This applies to both the outbound channel adapter and gateway. +NOTE +==== +Starting with _version 5.0_, headers that are added to the `MessageProperties` of the outbound message are never overwritten by mapped headers (by default). +Previously, this was only the case if the message converter was a `ContentTypeDelegatingMessageConverter` (in that case, the header was mapped first, so that the proper converter could be selected). +For other converters, such as the `SimpleMessageConverter`, mapped headers overwrote any headers added by the converter. +This caused problems when an outbound message had some left over `contentType` header (perhaps from an inbound channel adapter) and the correct outbound `contentType` was incorrectly overwritten. +The work-around was to use a header filter to remove the header before sending the message to the outbound endpoint. + +There are, however, cases where the previous behavior is desired. +For example, with a `String` payload containing JSON, the `SimpleMessageConverter` is not aware of the content and sets the `contentType` message property to `text/plain`, but your application would like to override that to `application/json` by setting the the `contentType` header of the message sent to the outbound endpoint. +The `ObjectToJsonTransformer` does exactly that (by default). + +There is now a property on the outbound channel adapter and gateway (as well as AMQP-backed channels) `headersMappedLast`. +Setting this to `true` will restore the behavior of overwriting the property added by the converter. +==== + [[amqp-user-id]] === Outbound User Id diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 629bd441e4..1fb3285d99 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -153,6 +153,10 @@ See <> for more information. Pollable AMQP-backed channels now block the poller thread for the poller's configured `receiveTimeout` (default 1 second). See <> for more information. +Headers, such as `contentType` that are added to message properties by the message converter are now used in the final message; previously, it depended on the converter type as to which headers/message properties appeared in the final message. +To override headers set by the converter, set the `headersMappedLast` property to `true`. +See <> for more information. + ==== HTTP Changes The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`.