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

INT-3361 Add AMQP DeliveryMode Attribute

Browse Source 
JIRA:https://jira.spring.io/browse/INT-3361

Add convenience attribute to simply configuration of
AMQP Delivery Mode (PERSISTENT, NON_PERSISTENT).

INT-3361: Polishing
This commit is contained in:
Gary Russell
2014-04-11 15:39:38 +03:00
committed by Artem Bilan
parent e0b567778f
commit 4577200e66
10 changed files with 214 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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.
@@ -16,6 +16,8 @@
package org.springframework.integration.amqp.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
@@ -24,7 +26,6 @@ import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for the AMQP 'outbound-channel-adapter' element.
@@ -48,6 +49,7 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key", true);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-delivery-mode");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -55,6 +55,8 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout", "sendTimeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-delivery-mode");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -17,6 +17,7 @@ import java.util.Map;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -81,43 +82,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
private volatile MessageChannel returnChannel;
@Override
protected void doInit() {
Assert.state(exchangeNameExpression == null || exchangeName == null,
"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
Assert.state(this.confirmCorrelationExpression == null || !this.expectReply,
"Confirm correlation expression does not apply to a gateway");
BeanFactory beanFactory = this.getBeanFactory();
if (exchangeNameExpression != null) {
Expression expression = expressionParser.parseExpression(this.exchangeNameExpression);
this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
if (beanFactory != null) {
this.exchangeNameGenerator.setBeanFactory(beanFactory);
}
}
Assert.state(routingKeyExpression == null || routingKey == null,
"Either a routingKey or a routingKeyExpression can be provided, but not both");
if (routingKeyExpression != null) {
Expression expression = expressionParser.parseExpression(this.routingKeyExpression);
this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
if (beanFactory != null) {
this.routingKeyGenerator.setBeanFactory(beanFactory);
}
}
if (this.confirmCorrelationExpression != null) {
Expression expression = expressionParser.parseExpression(this.confirmCorrelationExpression);
this.correlationDataGenerator = new ExpressionEvaluatingMessageProcessor<Object>(expression, Object.class);
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for publisher confirms");
((RabbitTemplate) this.amqpTemplate).setConfirmCallback(this);
if (beanFactory != null) {
this.correlationDataGenerator.setBeanFactory(beanFactory);
}
}
if (this.returnChannel != null) {
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for publisher returns");
((RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
}
}
private volatile MessageDeliveryMode defaultDeliveryMode;
public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
Assert.notNull(amqpTemplate, "amqpTemplate must not be null");
@@ -167,11 +132,55 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
this.returnChannel = returnChannel;
}
public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
this.defaultDeliveryMode = defaultDeliveryMode;
}
@Override
public String getComponentType() {
return expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter";
}
@Override
protected void doInit() {
Assert.state(exchangeNameExpression == null || exchangeName == null,
"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
Assert.state(this.confirmCorrelationExpression == null || !this.expectReply,
"Confirm correlation expression does not apply to a gateway");
BeanFactory beanFactory = this.getBeanFactory();
if (exchangeNameExpression != null) {
Expression expression = expressionParser.parseExpression(this.exchangeNameExpression);
this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
if (beanFactory != null) {
this.exchangeNameGenerator.setBeanFactory(beanFactory);
}
}
Assert.state(routingKeyExpression == null || routingKey == null,
"Either a routingKey or a routingKeyExpression can be provided, but not both");
if (routingKeyExpression != null) {
Expression expression = expressionParser.parseExpression(this.routingKeyExpression);
this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
if (beanFactory != null) {
this.routingKeyGenerator.setBeanFactory(beanFactory);
}
}
if (this.confirmCorrelationExpression != null) {
Expression expression = expressionParser.parseExpression(this.confirmCorrelationExpression);
this.correlationDataGenerator = new ExpressionEvaluatingMessageProcessor<Object>(expression, Object.class);
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
"RabbitTemplate implementation is required for publisher confirms");
((RabbitTemplate) this.amqpTemplate).setConfirmCallback(this);
if (beanFactory != null) {
this.correlationDataGenerator.setBeanFactory(beanFactory);
}
}
if (this.returnChannel != null) {
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
"RabbitTemplate implementation is required for publisher confirms");
((RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
}
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
String exchangeName = this.exchangeName;
@@ -209,9 +218,11 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
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;
}
},
@@ -220,6 +231,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
else {
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());
@@ -230,11 +242,13 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
private Message<?> sendAndReceive(String exchangeName, String routingKey, Message<?> requestMessage) {
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for send and receive");
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 amqpReplyMessage = this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage);
if (amqpReplyMessage == null) {
return null;
@@ -248,6 +262,14 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
return builder.build();
}
private void checkDeliveryMode(Message<?> requestMessage, MessageProperties messageProperties) {
if (this.defaultDeliveryMode != null &&
requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) {
messageProperties.setDeliveryMode(this.defaultDeliveryMode);
}
}
@Override
public void confirm(CorrelationData correlationData, boolean ack) {
Object userCorrelationData = correlationData;
if (correlationData instanceof CorrelationDataWrapper) {
@@ -271,21 +293,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
}
private static class CorrelationDataWrapper extends CorrelationData {
private final Object userData;
public CorrelationDataWrapper(String id, Object userData) {
super(id);
this.userData = userData;
}
public Object getUserData() {
return userData;
}
}
@Override
public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
String exchange, String routingKey) {
// safe to cast; we asserted we have a RabbitTemplate in doInit()
@@ -302,4 +310,21 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey);
this.returnChannel.send(builder.build());
}
private static class CorrelationDataWrapper extends CorrelationData {
private final Object userData;
public CorrelationDataWrapper(String id, Object userData) {
super(id);
this.userData = userData;
}
public Object getUserData() {
return this.userData;
}
}
}

25
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.0.xsd
View File

@@ -464,8 +464,8 @@ standard headers to also be mapped.
<xsd:attribute name="return-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Channel to which returned messages will be sent. Requires a RabbitTemplate with the 'mandatory' or
'immediate' properties set to TRUE; requires a CachingConnectionFactory with the 'publisherReturns'
Channel to which returned messages will be sent. Requires a RabbitTemplate with the 'mandatory'
property set to TRUE; requires a CachingConnectionFactory with the 'publisherReturns'
property set to TRUE.
]]></xsd:documentation>
<xsd:appinfo>
@@ -475,8 +475,29 @@ property set to TRUE.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="default-delivery-mode">
<xsd:annotation>
<xsd:documentation>
The default delivery mode for messages; 'PERSISTENT' or 'NON_PERSISTENT'. Overridden if the 'header-mapper'
sets the delivery mode. The 'DefaultHeaderMapper' sets the value if the
Spring Integration message header 'amqp_deliveryMode' is present. If this attribute is not supplied and
the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter'
used by the 'RabbitTemplate'. If that is not customized at all, the default is 'PERSISTENT'.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="deliveryModeEnumeration xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
<xsd:simpleType name="deliveryModeEnumeration">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="PERSISTENT" />
<xsd:enumeration value="NON_PERSISTENT" />
</xsd:restriction>
</xsd:simpleType>
<xsd:complexType name="inboundType">
<xsd:annotation>
<xsd:documentation>

1
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
View File

@@ -23,6 +23,7 @@
<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel"
exchange-name="outboundchanneladapter.test.1"
default-delivery-mode="NON_PERSISTENT"
mapped-request-headers="foo*"/>
<bean id="customHeaderMapper" class="org.mockito.Mockito" factory-method="mock">

28
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
View File

@@ -30,6 +30,7 @@ import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -37,6 +38,7 @@ import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -65,6 +67,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
@@ -82,6 +85,7 @@ import com.rabbitmq.client.Channel;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class AmqpOutboundChannelAdapterParserTests {
private static volatile int adviceCalled;
@@ -111,21 +115,25 @@ public class AmqpOutboundChannelAdapterParserTests {
Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomHeaders");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
final AtomicBoolean shouldBePersistent = new AtomicBoolean();
Mockito.doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpReplyMessage.getMessageProperties();
org.springframework.amqp.core.Message amqpMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
assertEquals("foobar", properties.getHeaders().get("foobar"));
assertNull(properties.getHeaders().get("bar"));
assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT
: MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode());
return null;
}
})
@@ -138,6 +146,15 @@ public class AmqpOutboundChannelAdapterParserTests {
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class));
shouldBePersistent.set(true);
message = MessageBuilder.withPayload("hello")
.setHeader("foo", "foo")
.setHeader("bar", "bar")
.setHeader("foobar", "foobar")
.setHeader(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.PERSISTENT)
.build();
requestChannel.send(message);
}
@Test
@@ -181,6 +198,7 @@ public class AmqpOutboundChannelAdapterParserTests {
List chainHandlers = TestUtils.getPropertyValue(eventDrivernConsumer, "handler.handlers", List.class);
AmqpOutboundEndpoint endpoint = (AmqpOutboundEndpoint) chainHandlers.get(0);
assertNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
@@ -191,8 +209,10 @@ public class AmqpOutboundChannelAdapterParserTests {
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2];
assertEquals("hello", new String(amqpReplyMessage.getBody()));
org.springframework.amqp.core.Message amqpMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpMessage.getMessageProperties();
assertEquals("hello", new String(amqpMessage.getBody()));
assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode());
return null;
}
})

1
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml
View File

@@ -40,6 +40,7 @@
routing-key="si.test.binding"
amqp-template="amqpTemplate"
order="5"
default-delivery-mode="NON_PERSISTENT"
requires-reply="false"
mapped-request-headers="foo*"
mapped-reply-headers="bar*"/>

58
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -24,24 +24,27 @@ import static org.junit.Assert.assertTrue;
import java.lang.reflect.Field;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.ReflectionUtils;
/**
@@ -59,7 +62,8 @@ public class AmqpOutboundGatewayParserTests {
@Test
public void testGatewayConfig(){
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object edc = context.getBean("rabbitGateway");
AmqpOutboundEndpoint gateway = TestUtils.getPropertyValue(edc, "handler", AmqpOutboundEndpoint.class);
assertEquals(5, gateway.getOrder());
@@ -72,15 +76,18 @@ public class AmqpOutboundGatewayParserTests {
Long sendTimeout = TestUtils.getPropertyValue(gateway, "messagingTemplate.sendTimeout", Long.class);
assertEquals(Long.valueOf(777), sendTimeout);
context.close();
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperCustomRequestResponse() {
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomRequestResponse");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
assertFalse(TestUtils.getPropertyValue(endpoint, "requiresReply", Boolean.class));
@@ -88,13 +95,17 @@ public class AmqpOutboundGatewayParserTests {
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
final AtomicBoolean shouldBePersistent = new AtomicBoolean();
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT
: MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode());
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
@@ -115,21 +126,35 @@ public class AmqpOutboundGatewayParserTests {
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
Message<?> replyMessage = queueChannel.receive(0);
assertNotNull(replyMessage);
assertEquals("bar", replyMessage.getHeaders().get("bar"));
assertEquals("foo", replyMessage.getHeaders().get("foo")); // copied from request Message
assertNull(replyMessage.getHeaders().get("foobar"));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
shouldBePersistent.set(true);
message = MessageBuilder.withPayload("hello")
.setHeader("foo", "foo")
.setHeader(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.PERSISTENT)
.build();
requestChannel.send(message);
replyMessage = queueChannel.receive(0);
assertNotNull(replyMessage);
context.close();
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperCustomAndStandardResponse() {
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
assertNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
@@ -137,7 +162,8 @@ public class AmqpOutboundGatewayParserTests {
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
@@ -147,6 +173,7 @@ public class AmqpOutboundGatewayParserTests {
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode());
org.springframework.amqp.core.Message amqpReplyMessage = new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
return amqpReplyMessage;
}})
@@ -168,12 +195,14 @@ public class AmqpOutboundGatewayParserTests {
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
context.close();
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperNothingToMap() {
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperNothingToMap");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
@@ -184,7 +213,8 @@ public class AmqpOutboundGatewayParserTests {
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
@@ -216,11 +246,13 @@ public class AmqpOutboundGatewayParserTests {
assertNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
assertEquals(1, adviceCalled);
context.close();
}
@Test //INT-1029
public void amqpOutboundGatewayWithinChain() {
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivenConsumer = context.getBean("chainWithRabbitOutboundGateway");
List<?> chainHandlers = TestUtils.getPropertyValue(eventDrivenConsumer, "handler.handlers", List.class);
@@ -233,6 +265,7 @@ public class AmqpOutboundGatewayParserTests {
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer<org.springframework.amqp.core.Message>() {
@Override
public org.springframework.amqp.core.Message answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
@@ -265,6 +298,7 @@ public class AmqpOutboundGatewayParserTests {
assertNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
context.close();
}

22
src/reference/docbook/amqp.xml
View File

@@ -294,6 +294,7 @@ this list can also be simple patterns to be matched against the header names (e.
order="1"]]><co id="amqp-outbound-channel-adapter-xml-5-co" linkends="amqp-outbound-channel-adapter-xml-5" /><![CDATA[
routing-key=""]]><co id="amqp-outbound-channel-adapter-xml-6-co" linkends="amqp-outbound-channel-adapter-xml-6" /><![CDATA[
routing-key-expression=""]]><co id="amqp-outbound-channel-adapter-xml-7-co" linkends="amqp-outbound-channel-adapter-xml-7" /><![CDATA[
default-delivery-mode""]]><co id="amqp-outbound-channel-adapter-xml-7a-co" linkends="amqp-outbound-channel-adapter-xml-7a" /><![CDATA[
confirm-correlation-expression=""]]><co id="amqp-outbound-channel-adapter-xml-8-co" linkends="amqp-outbound-channel-adapter-xml-8" /><![CDATA[
confirm-ack-channel=""]]><co id="amqp-outbound-channel-adapter-xml-9-co" linkends="amqp-outbound-channel-adapter-xml-9" /><![CDATA[
confirm-nack-channel=""]]><co id="amqp-outbound-channel-adapter-xml-10-co" linkends="amqp-outbound-channel-adapter-xml-10" /><![CDATA[
@@ -338,6 +339,16 @@ this list can also be simple patterns to be matched against the header names (e.
'payload.key'). By default, this will be an empty String.
<emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="amqp-outbound-channel-adapter-xml-7a-co" id="amqp-outbound-channel-adapter-xml-7a">
<para>
The default delivery mode for messages; 'PERSISTENT' or 'NON_PERSISTENT'. Overridden if the 'header-mapper'
sets the delivery mode. The 'DefaultHeaderMapper' sets the value if the
Spring Integration message header <code>amqp_deliveryMode</code>
is present. If this attribute is not supplied and
the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter'
used by the 'RabbitTemplate'. If that is not customized at all, the default is 'PERSISTENT'.
<emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="amqp-outbound-channel-adapter-xml-8-co" id="amqp-outbound-channel-adapter-xml-8">
<para>An expression defining correlation data. When provided, this configures the underlying
amqp template to receive publisher confirms. Requires a <classname>RabbitTemplate</classname> and a
@@ -446,6 +457,7 @@ this list can also be simple patterns to be matched against the header names (e.
reply-channel=""]]><co id="amqp-outbound-gateway-adapter-xml-6-co" linkends="amqp-outbound-gateway-adapter-xml-6" /><![CDATA[
routing-key=""]]><co id="amqp-outbound-gateway-adapter-xml-7-co" linkends="amqp-outbound-gateway-adapter-xml-7" /><![CDATA[
routing-key-expression=""]]><co id="amqp-outbound-gateway-adapter-xml-8-co" linkends="amqp-outbound-gateway-adapter-xml-8" /><![CDATA[
default-delivery-mode""]]><co id="amqp-outbound-gateway-adapter-xml-8a-co" linkends="amqp-outbound-channel-adapter-xml-8a" /><![CDATA[
return-channel=""]]><co id="amqp-outbound-gateway-adapter-xml-9-co" linkends="amqp-outbound-gateway-adapter-xml-9" /><![CDATA[/>]]>
</programlisting>
<para>
@@ -492,6 +504,16 @@ this list can also be simple patterns to be matched against the header names (e.
By default, this will be an empty String.
<emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="amqp-outbound-gateway-adapter-xml-8a-co" id="amqp-outbound-gateway-adapter-xml-7a">
<para>
The default delivery mode for messages; 'PERSISTENT' or 'NON_PERSISTENT'. Overridden if the 'header-mapper'
sets the delivery mode. The 'DefaultHeaderMapper' sets the value if the
Spring Integration message header <code>amqp_deliveryMode</code>
is present. If this attribute is not supplied and
the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter'
used by the 'RabbitTemplate'. If that is not customized at all, the default is 'PERSISTENT'.
<emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="amqp-outbound-gateway-adapter-xml-9-co" id="amqp-outbound-gateway-adapter-xml-9">
<para>The channel to which returned messages are sent. When provided, the underlying amqp template
is configured to return undeliverable messages to the gateway. The message will be constructed from the

11
src/reference/docbook/whats-new.xml
View File

@@ -271,5 +271,16 @@
For more information, see <xref linkend="jdbc-message-store-channels"/>.
</para>
</section>
<section id="4.0-amqp">
<title>AMQP Endpoints Delivery Mode</title>
<para>
Spring AMQP, by default, creates persistent messages on the broker. This behavior can be
overridden by setting the <code>amqp_deliveryMode</code> header and/or customizing the
mappers. A convenient <code>default-delivery-mode</code> attribute has now been added
to the adapters to provide easier configuration of this important setting.
For more information, see <xref linkend="amqp-outbound-channel-adapter"/> and
<xref linkend="amqp-outbound-gateway"/>.
</para>
</section>
</section>
</chapter>