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 71fb21b085..56fb4b4463 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
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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
@@ -27,6 +27,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationListener;
@@ -340,18 +341,9 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
private void send(String exchangeName, String routingKey,
final Message requestMessage, CorrelationData correlationData) {
if (this.amqpTemplate instanceof RabbitTemplate) {
- ((RabbitTemplate) this.amqpTemplate).convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
- new MessagePostProcessor() {
- @Override
- public org.springframework.amqp.core.Message postProcessMessage(
- org.springframework.amqp.core.Message message) throws AmqpException {
- headerMapper.fromHeadersToRequest(requestMessage.getHeaders(),
- message.getMessageProperties());
- checkDeliveryMode(requestMessage, message.getMessageProperties());
- return message;
- }
- },
- correlationData);
+ MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
+ org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter);
+ ((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData);
}
else {
this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
@@ -372,13 +364,9 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
"RabbitTemplate implementation is required for publisher confirms");
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
- MessageProperties amqpMessageProperties = new MessageProperties();
- org.springframework.amqp.core.Message amqpMessage =
- converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
- this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
- checkDeliveryMode(requestMessage, amqpMessageProperties);
+ org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter);
org.springframework.amqp.core.Message amqpReplyMessage =
- ((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey,amqpMessage,
+ ((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage,
correlationData);
if (amqpReplyMessage == null) {
@@ -393,6 +381,21 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
return builder.build();
}
+ protected org.springframework.amqp.core.Message mapMessage(Message requestMessage, MessageConverter converter) {
+ MessageProperties amqpMessageProperties = new MessageProperties();
+ org.springframework.amqp.core.Message amqpMessage;
+ if (converter instanceof ContentTypeDelegatingMessageConverter) {
+ this.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);
+ this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
+ }
+ checkDeliveryMode(requestMessage, amqpMessageProperties);
+ return amqpMessage;
+ }
+
private void checkDeliveryMode(Message requestMessage, MessageProperties messageProperties) {
if (this.defaultDeliveryMode != null &&
requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) {
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml
index 6b2eaf2918..3fd372132c 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml
@@ -13,14 +13,15 @@
-
+
-
-
+
@@ -73,11 +74,31 @@
-
+
+
+
+
+
+
+
+
+
+
+
+
-
+
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
index 2ef609ef53..16185de12b 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -25,10 +25,12 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.amqp.rule.BrokerRunning;
+import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -71,7 +73,7 @@ public class AmqpOutboundEndpointTests {
private PollableChannel ackChannel;
@Autowired
- private MessageChannel pcRequestChannelAdapter;
+ private MessageChannel pcRequestChannelForAdapter;
@Autowired
private MessageChannel returnRequestChannel;
@@ -79,11 +81,21 @@ public class AmqpOutboundEndpointTests {
@Autowired
private PollableChannel returnChannel;
+ @Autowired
+ private MessageChannel ctRequestChannel;
+
+ @Autowired
+ private ConnectionFactory connectionFactory;
+
@Test
public void testGatewayPublisherConfirms() throws Exception {
+ while (this.amqpTemplateConfirms.receive(this.queue.getName()) != null) {
+ ;
+ }
Message message = MessageBuilder.withPayload("hello")
.setHeader("amqp_confirmCorrelationData", "foo")
+ .setHeader(AmqpHeaders.CONTENT_TYPE, "application/json")
.build();
this.pcRequestChannel.send(message);
Message ack = this.ackChannel.receive(10000);
@@ -91,6 +103,12 @@ public class AmqpOutboundEndpointTests {
assertEquals("foo", ack.getPayload());
assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
+ org.springframework.amqp.core.Message received = this.amqpTemplateConfirms.receive(this.queue.getName());
+ assertEquals("\"hello\"", new String(received.getBody(), "UTF-8"));
+ assertEquals("application/json", received.getMessageProperties().getContentType());
+ assertEquals("java.lang.String", received.getMessageProperties().getHeaders()
+ .get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")));
+
// test whole message is correlation
message = MessageBuilder.withPayload("hello")
.build();
@@ -100,8 +118,9 @@ public class AmqpOutboundEndpointTests {
assertSame(message.getPayload(), ack.getPayload());
assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
- this.amqpTemplateConfirms.receive(this.queue.getName()); // so queue is deleted
-
+ while (this.amqpTemplateConfirms.receive(this.queue.getName()) != null) {
+ ;
+ }
}
@Test
@@ -109,7 +128,7 @@ public class AmqpOutboundEndpointTests {
Message message = MessageBuilder.withPayload("hello")
.setHeader("amqp_confirmCorrelationData", "foo")
.build();
- this.pcRequestChannelAdapter.send(message);
+ this.pcRequestChannelForAdapter.send(message);
Message ack = this.ackChannel.receive(10000);
assertNotNull(ack);
assertEquals("foo", ack.getPayload());
@@ -125,5 +144,33 @@ public class AmqpOutboundEndpointTests {
assertEquals(message.getPayload(), returned.getPayload());
}
+ @Test
+ public void adapterWithContentType() throws Exception {
+ RabbitTemplate template = new RabbitTemplate(this.connectionFactory);
+ template.setQueue(this.queue.getName());
+ while (template.receive() != null) {
+ ;
+ }
+ Message message = MessageBuilder.withPayload("hello")
+ .setHeader(AmqpHeaders.CONTENT_TYPE, "application/json")
+ .build();
+ this.ctRequestChannel.send(message);
+ org.springframework.amqp.core.Message m = template.receive();
+ assertNotNull(m);
+ assertEquals("\"hello\"", new String(m.getBody(), "UTF-8"));
+ assertEquals("application/json", m.getMessageProperties().getContentType());
+ assertEquals("java.lang.String",
+ m.getMessageProperties().getHeaders().get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")));
+ message = MessageBuilder.withPayload("hello")
+ .build();
+ this.ctRequestChannel.send(message);
+ m = template.receive();
+ assertNotNull(m);
+ assertEquals("hello", new String(m.getBody(), "UTF-8"));
+ assertEquals("text/plain", m.getMessageProperties().getContentType());
+ while (template.receive() != null) {
+ ;
+ }
+ }
}
diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc
index 165193000d..f235f2aa5f 100644
--- a/src/reference/asciidoc/amqp.adoc
+++ b/src/reference/asciidoc/amqp.adoc
@@ -906,6 +906,48 @@ public class AmqpJavaApplication {
}
----
+[[content-type-conversion-outbound]]
+=== Outbound Message Conversion
+
+Spring AMQP 1.4 introduced the `ContentTypeDelegatingMessageConverter` where the actual converter is selected based
+on the incoming content type message property.
+This could be used by inbound endpoints.
+
+Spring Integration _version 4.3_ now allows the `ContentTypeDelegatingMessageConverter` to be used on outbound
+endpoints as well - with the `contentType` header specifiying which converter will be used.
+
+The following configures a `ContentTypeDelegatingMessageConverter` with the default converter being the
+`SimpleMessageConverter` (which handles java serialization and plain text), together with a JSON converter:
+
+[source, xml]
+----
+
+
+
+
+
+
+
+
+
+
+
+----
+
+Sending a message to `ctRequestChannel` with the `contentType` header set to `application/json` will cause the
+JSON converter to be selected.
+
+This applies to both the outbound channel adapter and gateway.
+
[[amqp-channels]]
=== AMQP Backed Message Channels
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc
index bf1e39953d..e6bd77b48e 100644
--- a/src/reference/asciidoc/whats-new.adoc
+++ b/src/reference/asciidoc/whats-new.adoc
@@ -52,3 +52,9 @@ See <> for more information.
The generated file name for the `FileWritingMessageHandler` can represent _sub-path_ to save the desired directory
structure for file in the target directory.
See <> for more information.
+
+==== AMQP Changes
+
+The outbound endpoints now support a `RabbitTemplate` configured with a `ContentTypeDelegatingMessageConverter` such
+that the converter can be chosen based on the message content type.
+See <> for more information.