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. + +