Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3890: AMQP: Content-Type Conversion Outbound

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3890
This commit is contained in:
Gary Russell
2016-01-05 16:10:13 -05:00
committed by Artem Bilan
parent fec2a36f42
commit 11ba07e3e3
5 changed files with 149 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
View File

@@ -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) {

33
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml
View File

@@ -13,14 +13,15 @@
<int:queue />
</int:channel>
<rabbit:template id="amqpTemplateConfirms" connection-factory="connectionFactory" reply-timeout="10" />
<rabbit:template id="amqpTemplateConfirms" connection-factory="connectionFactory"
reply-timeout="10" message-converter="ctConverter" />
<amqp:outbound-gateway id="pcGateway"
request-channel="pcRequestChannel"
reply-channel="fromRabbit"
exchange-name=""
routing-key="#{queue.name}"
mapped-request-headers="foo*"
mapped-request-headers="STANDARD_REQUEST_HEADERS,foo*"
requires-reply="false"
amqp-template="amqpTemplateConfirms"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
@@ -48,7 +49,7 @@
<rabbit:template id="amqpTemplateConfirmsAdapter" connection-factory="connectionFactory"/>
<amqp:outbound-channel-adapter id="withPublisherConfirms" channel="pcRequestChannelAdapter"
<amqp:outbound-channel-adapter id="withPublisherConfirms" channel="pcRequestChannelForAdapter"
exchange-name=""
routing-key="#{queue.name}"
mapped-request-headers="foo*"
@@ -56,7 +57,7 @@
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel" />
<int:channel id="pcRequestChannelAdapter"/>
<int:channel id="pcRequestChannelForAdapter"/>
<rabbit:template id="amqpTemplateReturns" connection-factory="connectionFactory" mandatory="true" />
@@ -73,11 +74,31 @@
<int:queue />
</int:channel>
<rabbit:connection-factory id="connectionFactory"
<amqp:outbound-channel-adapter id="withContentTypeConverter" channel="ctRequestChannel"
exchange-name=""
routing-key="#{queue.name}"
amqp-template="amqpTemplateContentTypeConverter" />
<int:channel id="ctRequestChannel"/>
<rabbit:template id="amqpTemplateContentTypeConverter"
connection-factory="connectionFactory" message-converter="ctConverter" />
<bean id="ctConverter" class="org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter">
<property name="delegates">
<map>
<entry key="application/json">
<bean class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
</entry>
</map>
</property>
</bean>
<rabbit:connection-factory id="connectionFactory"
host="localhost" publisher-returns="true" publisher-confirms="true" />
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:queue id="queue" />
</beans>

57
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
View File

@@ -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) {
;
}
}
}

42
src/reference/asciidoc/amqp.adoc
View File

@@ -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]
----
<amqp:outbound-channel-adapter id="withContentTypeConverter" channel="ctRequestChannel"
exchange-name="someExchange"
routing-key="someKey"
amqp-template="amqpTemplateContentTypeConverter" />
<int:channel id="ctRequestChannel"/>
<rabbit:template id="amqpTemplateContentTypeConverter"
connection-factory="connectionFactory" message-converter="ctConverter" />
<bean id="ctConverter"
class="o.s.amqp.support.converter.ContentTypeDelegatingMessageConverter">
<property name="delegates">
<map>
<entry key="application/json">
<bean class="o.s.amqp.support.converter.Jackson2JsonMessageConverter" />
</entry>
</map>
</property>
</bean>
----
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

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -52,3 +52,9 @@ See <<udp-adapters>> 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 <<file-writing-file-names>> 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 <<content-type-conversion-outbound>> for more information.