From 81bfaeb8a9761b482f9bf9924d6fd04a7b10e612 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 2 May 2013 12:18:27 -0400 Subject: [PATCH] INT-3002 Fix Content-Type Mapping https://jira.springsource.org/browse/INT-3002 The AMQP outbound endpoint behaves differently when configured as a gateway compared to when it is configured an adapter. With the adapter, the header mapper is invoked after the message converter (via the MPP callback). With the gateway, the mapper is invoked before the converter. The SimpleMessageConverter unconditionally sets the content-type, overriding any mapped content type. Change the code to invoke the header mapper after the converter to match the adapter logic. Remove the TODO. The method now exists, but Using a MPP in the gateway case does not provide access to headers on the reply. Add test cases. --- .../amqp/outbound/AmqpOutboundEndpoint.java | 3 +- .../amqp/outbound/OutboundEndpointTests.java | 111 ++++++++++++++++++ src/reference/docbook/whats-new.xml | 18 +++ 3 files changed, 130 insertions(+), 2 deletions(-) create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java 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 d204fedb78..5387b7c2e6 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 @@ -231,12 +231,11 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler } private Message sendAndReceive(String exchangeName, String routingKey, Message requestMessage) { - // TODO: add a convertSendAndReceive method that accepts a MessagePostProcessor so we can map headers? Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for send and receive"); MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); MessageProperties amqpMessageProperties = new MessageProperties(); - this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); org.springframework.amqp.core.Message amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); + this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); org.springframework.amqp.core.Message amqpReplyMessage = this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage); if (amqpReplyMessage == null) { return null; 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 new file mode 100644 index 0000000000..d78a57d224 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java @@ -0,0 +1,111 @@ +/* + * Copyright 2002-2013 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.outbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; + +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.support.CorrelationData; +import org.springframework.integration.MessageHeaders; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Gary Russell + * @since 3.0 + */ +public class OutboundEndpointTests { + + @Test + public void testHeaderMapperWinsAdapter() { + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + RabbitTemplate amqpTemplate = spy(new RabbitTemplate(connectionFactory)); + AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate); + final AtomicReference amqpMessage = + new AtomicReference(); + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + amqpMessage.set((Message) invocation.getArguments()[2]); + return null; + } + }).when(amqpTemplate).send(anyString(), anyString(), any(Message.class), + any(CorrelationData.class)); + org.springframework.integration.Message message = MessageBuilder.withPayload("foo") + .setHeader(MessageHeaders.CONTENT_TYPE, "bar") + .build(); + endpoint.handleMessage(message); + assertNotNull(amqpMessage.get()); + assertEquals("bar", amqpMessage.get().getMessageProperties().getContentType()); + } + + @Test + public void testHeaderMapperWinsGateway() { + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + TestRabbitTemplate amqpTemplate = spy(new TestRabbitTemplate(connectionFactory)); + AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate); + endpoint.setExpectReply(true); + final AtomicReference amqpMessage = + new AtomicReference(); + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + amqpMessage.set((Message) invocation.getArguments()[2]); + return null; + } + }).when(amqpTemplate).doSendAndReceiveWithTemporary(anyString(), anyString(), any(Message.class)); + org.springframework.integration.Message message = MessageBuilder.withPayload("foo") + .setHeader(MessageHeaders.CONTENT_TYPE, "bar") + .build(); + endpoint.handleMessage(message); + assertNotNull(amqpMessage.get()); + assertEquals("bar", amqpMessage.get().getMessageProperties().getContentType()); + } + + /** + * Increase method visibility + */ + private class TestRabbitTemplate extends RabbitTemplate { + + private TestRabbitTemplate(ConnectionFactory connectionFactory) { + super(connectionFactory); + } + + @Override + public org.springframework.amqp.core.Message doSendAndReceiveWithTemporary(String exchange, + String routingKey, org.springframework.amqp.core.Message message) { + return super.doSendAndReceiveWithTemporary(exchange, routingKey, message); + } + + } + +} diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index f1e9672cbe..66cb225ec4 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -144,6 +144,24 @@ before sending the request. For more information see . +
+ AMQP Outbound Gateway Header Mapping + + Previously, the <int-amqp:outbound-gateway/> mapped headers before invoking the message + converter, and the converter could overwrite headers such as content-type. The + outbound adapter maps the headers after the conversion, which means headers like + content-type from the outbound Message (if present) are used. + + + Starting with this release, the gateway now maps the headers after the message conversion, + consistent with the adapter. If your application relies on the previous behavior (where the + converter's headers overrode the mapped headers), you either need to filter those headers + (before the message reaches the gateway) + or set them appropriately. The headers affected by the SimpleMessageConverter + are content-type and content-encoding. Custom message converters + may set other headers. + +