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.

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;

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<Message> amqpMessage =
+				new AtomicReference<Message>();
+		doAnswer(new Answer<Object>() {
+
+			@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<Message> amqpMessage =
+				new AtomicReference<Message>();
+		doAnswer(new Answer<Object>() {
+
+			@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);
+		}
+
+	}
+
+}

src/reference/docbook/whats-new.xml

@@ -144,6 +144,24 @@
 				before sending the request. For more information see <xref linkend="http"/>.
 			</para>
 		</section>
+		<section id="3.0-amqp-mapping">
+			<title>AMQP Outbound Gateway Header Mapping</title>
+			<para>
+				Previously, the &lt;int-amqp:outbound-gateway/&gt; mapped headers before invoking the message
+				converter, and the converter could overwrite headers such as <code>content-type</code>. The
+				outbound adapter maps the headers after the conversion, which means headers like
+				<code>content-type</code> from the outbound <code>Message</code> (if present) are used.
+			</para>
+			<para>
+				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 <classname>SimpleMessageConverter</classname>
+				are <code>content-type</code> and <code>content-encoding</code>. Custom message converters
+				may set other headers.
+			</para>
+		</section>
 
 	</section>
 </chapter>