From 4577200e66f89d6e416a551844b3824e292a5fb6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 11 Apr 2014 15:39:38 +0300 Subject: [PATCH] INT-3361 Add AMQP DeliveryMode Attribute JIRA:https://jira.spring.io/browse/INT-3361 Add convenience attribute to simply configuration of AMQP Delivery Mode (PERSISTENT, NON_PERSISTENT). INT-3361: Polishing --- .../AmqpOutboundChannelAdapterParser.java | 6 +- .../config/AmqpOutboundGatewayParser.java | 4 +- .../amqp/outbound/AmqpOutboundEndpoint.java | 133 +++++++++++------- .../config/spring-integration-amqp-4.0.xsd | 25 +++- ...boundChannelAdapterParserTests-context.xml | 1 + ...AmqpOutboundChannelAdapterParserTests.java | 28 +++- ...AmqpOutboundGatewayParserTests-context.xml | 1 + .../AmqpOutboundGatewayParserTests.java | 58 ++++++-- src/reference/docbook/amqp.xml | 22 +++ src/reference/docbook/whats-new.xml | 11 ++ 10 files changed, 214 insertions(+), 75 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java index 601d9822ea..8c37ca3771 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java @@ -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); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java index 58bc7315ba..7e51d7b3bb 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java @@ -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"); 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 4b38b436f8..ace6a68442 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-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(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(expression, String.class); - if (beanFactory != null) { - this.routingKeyGenerator.setBeanFactory(beanFactory); - } - } - if (this.confirmCorrelationExpression != null) { - Expression expression = expressionParser.parseExpression(this.confirmCorrelationExpression); - this.correlationDataGenerator = new ExpressionEvaluatingMessageProcessor(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(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(expression, String.class); + if (beanFactory != null) { + this.routingKeyGenerator.setBeanFactory(beanFactory); + } + } + if (this.confirmCorrelationExpression != null) { + Expression expression = expressionParser.parseExpression(this.confirmCorrelationExpression); + this.correlationDataGenerator = new ExpressionEvaluatingMessageProcessor(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; + } + + } + } diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.0.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.0.xsd index 446ea49d02..0cf9241d67 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.0.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.0.xsd @@ -464,8 +464,8 @@ standard headers to also be mapped. @@ -475,8 +475,29 @@ property set to TRUE. + + + + 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'. + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml index daf07d34ad..96c79e8da1 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml @@ -23,6 +23,7 @@ diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java index 1d2065f54b..2a417c6817 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java @@ -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() { @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; } }) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml index f42ad1aefc..add67812cc 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml @@ -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*"/> diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index 786801b4a2..2b3706107e 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -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() { + @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(); } diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index a487da810e..798b118295 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -294,6 +294,7 @@ this list can also be simple patterns to be matched against the header names (e. order="1"]]>Optional. + + + 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'. + Optional. + An expression defining correlation data. When provided, this configures the underlying amqp template to receive publisher confirms. Requires a RabbitTemplate and a @@ -446,6 +457,7 @@ this list can also be simple patterns to be matched against the header names (e. reply-channel=""]]>]]> @@ -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. Optional. + + + 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'. + Optional. + 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 diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 8196f1f57c..6ba9b4b459 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -271,5 +271,16 @@ For more information, see . +
+ AMQP Endpoints Delivery Mode + + Spring AMQP, by default, creates persistent messages on the broker. This behavior can be + overridden by setting the amqp_deliveryMode header and/or customizing the + mappers. A convenient default-delivery-mode attribute has now been added + to the adapters to provide easier configuration of this important setting. + For more information, see and + . + +