From 8b5c1c5053d37ffd6a01629322110426d730446e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 19 Aug 2014 10:40:11 -0400 Subject: [PATCH] INT-3499 AMQP Confirms/Returns Polishing JIRA: https://jira.spring.io/browse/INT-3499 - Suppress expected exception from log in test case - Suppress (log) 'null' payload error when no correlation data - Enforce a specific amqp template reference in the parser when using confirms/returns - Change tests to use a dedicated template for confirms/returns --- .../AmqpOutboundChannelAdapterParser.java | 5 +++++ .../amqp/outbound/AmqpOutboundEndpoint.java | 6 ++++++ .../config/spring-integration-amqp-4.1.xsd | 2 +- ...boundChannelAdapterParserTests-context.xml | 6 ++++++ ...AmqpOutboundChannelAdapterParserTests.java | 8 +++++--- src/reference/docbook/amqp.xml | 20 ++++++++++--------- 6 files changed, 34 insertions(+), 13 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 6f65114b0e..d1899cc825 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 @@ -43,6 +43,11 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda String amqpTemplateRef = element.getAttribute("amqp-template"); if (!StringUtils.hasText(amqpTemplateRef)) { amqpTemplateRef = "amqpTemplate"; + if (StringUtils.hasText(element.getAttribute("return-channel")) + || StringUtils.hasText(element.getAttribute("confirm-correlation-expression"))) { + parserContext.getReaderContext().error("A dedicated 'amqp-template' is required when" + + " using publisher confirms and returns", element); + } } builder.addConstructorArgReference(amqpTemplateRef); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name", true); 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 a8c51da644..31b4b3ea70 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 @@ -322,6 +322,12 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { Object userCorrelationData = correlationData; + if (correlationData == null) { + if (logger.isDebugEnabled()) { + logger.debug("No correlation data provided for ack: " + ack + " cause:" + cause); + } + return; + } if (correlationData instanceof CorrelationDataWrapper) { userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData(); } diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd index 788eb556a5..9bff535de7 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd @@ -698,7 +698,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA Flag to indicate that channels created by this component will be transactional. - Only applies to outbound messages when 'message-driven' is 'true'. + Only applies to messages received from this channel when 'message-driven' is 'true'. 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 2abb6c8574..7652342a50 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 @@ -41,9 +41,12 @@ + + @@ -53,9 +56,12 @@ + + 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 a192309c6f..33012e93b9 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 @@ -39,8 +39,6 @@ import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; -import com.rabbitmq.client.AMQP.BasicProperties; -import com.rabbitmq.client.Channel; import org.apache.commons.logging.Log; import org.junit.Test; import org.junit.runner.RunWith; @@ -86,6 +84,9 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ReflectionUtils; +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Channel; + /** * @author Mark Fisher * @author Oleg Zhurakousky @@ -256,7 +257,7 @@ public class AmqpOutboundChannelAdapterParserTests { Message message = MessageBuilder.withPayload("hello").build(); requestChannel.send(message); PollableChannel returnChannel = context.getBean("returnChannel", PollableChannel.class); - RabbitTemplate template = context.getBean("amqpTemplate", RabbitTemplate.class); + RabbitTemplate template = context.getBean("amqpTemplateReturns", RabbitTemplate.class); Map headers = new HashMap(); headers.put(PublisherCallbackChannel.RETURN_CORRELATION, template.getUUID()); BasicProperties properties = mock(BasicProperties.class); @@ -360,6 +361,7 @@ public class AmqpOutboundChannelAdapterParserTests { AmqpOutboundEndpoint handler = new AmqpOutboundEndpoint(amqpTemplate); Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class)); new DirectFieldAccessor(handler).setPropertyValue("logger", logger); + doAnswer(new DoesNothing()).when(logger).error("Failed to eagerly establish the connection.", toBeThrown); ApplicationContext context = mock(ApplicationContext.class); handler.setApplicationContext(context); handler.setBeanFactory(context); diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index 5ca9da1b29..052219fbc5 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -481,18 +481,20 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann An expression defining correlation data. When provided, this configures the underlying - amqp template to receive publisher confirms. Requires a RabbitTemplate and a + amqp template to receive publisher confirms. Requires a dedicated + RabbitTemplate and a CachingConnectionFactory with the publisherConfirms property set to true. When a publisher confirm - is received, it is written to either the confirm-ack-channel, or the confirm-nack-channel, + is received, and correlation data is supplied, + it is written to either the confirm-ack-channel, or the confirm-nack-channel, depending on the confirmation type. The payload of the confirm is the correlation data as defined by this expression and the message will have a header 'amqp_publishConfirm' set to true (ack) or false (nack). Examples: "headers['myCorrelationData']", "payload". Optional. - Starting with version 4.1 the new amqp_publishConfirmNackCause - message header has been added. It contains a cause message of 'nack' for publisher - confirm. + Starting with version 4.1 the amqp_publishConfirmNackCause + message header has been added. It contains the cause of a 'nack' for publisher + confirms. @@ -512,8 +514,8 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey. Optional. - Using a return-channel requires a RabbitTemplate with either - the mandatory or immediate properties set to true, + Using a return-channel requires a RabbitTemplate with + the mandatory property set to true, and a CachingConnectionFactory with the publisherReturns property set to true. When using multiple outbound endpoints with returns, a separate RabbitTemplate is needed @@ -643,8 +645,8 @@ public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Chann amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey. Optional. - Using a return-channel requires a RabbitTemplate with either - the mandatory or immediate properties set to true, + Using a return-channel requires a RabbitTemplate with + the mandatory property set to true, and a CachingConnectionFactory with the publisherReturns property set to true. When using multiple outbound endpoints with returns, a separate RabbitTemplate is needed