From 135287994b3d0d4c3dc39d43737779a095bb86c3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 28 Apr 2015 16:12:58 +0300 Subject: [PATCH] INT-3582: Pub Confirms for AMQP Outbound Gateway JIRA: https://jira.spring.io/browse/INT-3582 --- .../config/AmqpOutboundGatewayParser.java | 8 +- .../amqp/outbound/AmqpOutboundEndpoint.java | 13 +-- .../config/spring-integration-amqp-4.2.xsd | 86 +++++++++---------- ...AmqpOutboundGatewayParserTests-context.xml | 17 ++++ .../AmqpOutboundGatewayParserTests.java | 61 +++++++++++-- .../amqp/outbound/OutboundEndpointTests.java | 10 ++- src/reference/asciidoc/amqp.adoc | 34 ++++++-- src/reference/asciidoc/whats-new.adoc | 8 ++ 8 files changed, 166 insertions(+), 71 deletions(-) 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 318ffbe054..37d55941d5 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-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -61,8 +61,12 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-correlation-expression"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel"); - IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null); + IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, + null); return builder; } 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 94e9f69264..b31a1f2a11 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-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -213,8 +213,6 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler 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 (this.exchangeNameExpression != null) { this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor(this.exchangeNameExpression, @@ -317,7 +315,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler routingKey = this.routingKeyGenerator.processMessage(requestMessage); } if (this.expectReply) { - return this.sendAndReceive(exchangeName, routingKey, requestMessage); + return this.sendAndReceive(exchangeName, routingKey, requestMessage, correlationData); } else { this.send(exchangeName, routingKey, requestMessage, correlationData); @@ -355,7 +353,8 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler } } - private Message sendAndReceive(String exchangeName, String routingKey, Message requestMessage) { + private Message sendAndReceive(String exchangeName, String routingKey, Message requestMessage, + CorrelationData correlationData) { Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate, "RabbitTemplate implementation is required for publisher confirms"); MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); @@ -365,7 +364,9 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties); checkDeliveryMode(requestMessage, amqpMessageProperties); org.springframework.amqp.core.Message amqpReplyMessage = - this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage); + ((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey,amqpMessage, + correlationData); + if (amqpReplyMessage == null) { return null; } diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd index 8eafa33727..b32117690e 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd @@ -43,49 +43,6 @@ - - - - - - - - - - - - - - - - - - - - - - - - - @@ -501,6 +458,49 @@ property set to TRUE. + + + + + + + + + + + + + + + + + + + + + + + + + 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 afa225d3a0..f4f2c433ee 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 @@ -80,4 +80,21 @@ mapped-reply-headers=""/> + + + + + + + + + + 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 34e5b2a778..f41f700f66 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-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.amqp.config; import static org.junit.Assert.assertEquals; @@ -21,6 +22,11 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; import java.lang.reflect.Field; import java.util.List; @@ -28,12 +34,17 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import org.mockito.Mockito; +import org.mockito.internal.stubbing.answers.DoesNothing; 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; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.support.CorrelationData; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.ConfigurableApplicationContext; @@ -45,8 +56,11 @@ 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.messaging.PollableChannel; import org.springframework.util.ReflectionUtils; +import com.rabbitmq.client.Channel; + /** * @author Oleg Zhurakousky * @author Gary Russell @@ -119,7 +133,7 @@ public class AmqpOutboundGatewayParserTests { } }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), - Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); MessageChannel requestChannel = context.getBean("toRabbit1", MessageChannel.class); @@ -127,7 +141,8 @@ public class AmqpOutboundGatewayParserTests { requestChannel.send(message); Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), - Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), + any(CorrelationData.class)); // verify reply QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class); @@ -185,7 +200,7 @@ public class AmqpOutboundGatewayParserTests { } }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), - Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); MessageChannel requestChannel = context.getBean("toRabbit2", MessageChannel.class); @@ -193,7 +208,8 @@ public class AmqpOutboundGatewayParserTests { requestChannel.send(message); Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), - Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), + any(CorrelationData.class)); // verify reply QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class); @@ -238,7 +254,7 @@ public class AmqpOutboundGatewayParserTests { } }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), - Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); MessageChannel requestChannel = context.getBean("toRabbit3", MessageChannel.class); @@ -246,7 +262,8 @@ public class AmqpOutboundGatewayParserTests { requestChannel.send(message); Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), - Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), + any(CorrelationData.class)); // verify reply QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class); @@ -292,7 +309,7 @@ public class AmqpOutboundGatewayParserTests { } }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), - Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate); @@ -301,7 +318,8 @@ public class AmqpOutboundGatewayParserTests { requestChannel.send(message); Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), - Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class)); + Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), + any(CorrelationData.class)); // verify reply QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class); @@ -329,6 +347,31 @@ public class AmqpOutboundGatewayParserTests { } } + @Test + public void testPublisherConfirms() throws Exception { + ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( + "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); + ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(connectionFactory.createConnection()).thenReturn(mockConnection); + PublisherCallbackChannelImpl publisherCallbackChannel = spy(new PublisherCallbackChannelImpl(mockChannel)); + doAnswer(new DoesNothing()).when(publisherCallbackChannel).close(); + when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel); + + MessageChannel requestChannel = context.getBean("pcRequestChannel", MessageChannel.class); + Message message = MessageBuilder.withPayload("hello") + .setHeader("amqp_confirmCorrelationData", "foo") + .build(); + requestChannel.send(message); + PollableChannel ackChannel = context.getBean("ackChannel", PollableChannel.class); + publisherCallbackChannel.handleAck(0, false); + Message ack = ackChannel.receive(1000); + assertNotNull(ack); + assertEquals("foo", ack.getPayload()); + assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); + } public static class FooAdvice extends AbstractRequestHandlerAdvice { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java index 6a4bca564b..b74760964d 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -42,6 +42,7 @@ import org.springframework.messaging.MessageHeaders; /** * @author Gary Russell + * @author Artem Bilan * @since 3.0 */ public class OutboundEndpointTests { @@ -88,7 +89,8 @@ public class OutboundEndpointTests { amqpMessage.set((Message) invocation.getArguments()[2]); return null; } - }).when(amqpTemplate).doSendAndReceiveWithTemporary(anyString(), anyString(), any(Message.class)); + }).when(amqpTemplate) + .doSendAndReceiveWithTemporary(anyString(), anyString(), any(Message.class), any(CorrelationData.class)); org.springframework.messaging.Message message = MessageBuilder.withPayload("foo") .setHeader(MessageHeaders.CONTENT_TYPE, "bar") .setReplyChannel(new QueueChannel()) @@ -110,8 +112,8 @@ public class OutboundEndpointTests { @Override public org.springframework.amqp.core.Message doSendAndReceiveWithTemporary(String exchange, - String routingKey, org.springframework.amqp.core.Message message) { - return super.doSendAndReceiveWithTemporary(exchange, routingKey, message); + String routingKey, org.springframework.amqp.core.Message message, CorrelationData correlationData) { + return super.doSendAndReceiveWithTemporary(exchange, routingKey, message, correlationData); } } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 6769ed3298..bb99fcad84 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -381,9 +381,10 @@ _Optional_. <10> An expression defining correlation data. When provided, this configures the underlying 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, 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). +Requires a dedicated `RabbitTemplate` and a `CachingConnectionFactory` with the `publisherConfirms` property set to +`true`. When a publisher confirm 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 `amqp_publishConfirmNackCause` message header has been added. @@ -450,8 +451,11 @@ A configuration sample for an AMQP Outbound Gateway is shown below. routing-key="" <10> routing-key-expression="" <11> default-delivery-mode"" <12> - return-channel="" <13> - lazy-connect="true" /> <14> + confirm-correlation-expression="" <13> + confirm-ack-channel="" <14> + confirm-nack-channel="" <15> + return-channel="" <16> + lazy-connect="true" /> <17> ---- @@ -512,15 +516,31 @@ The 'DefaultHeaderMapper' sets the value if the Spring Integration message heade 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_. +<13> Since _version 4.2_. An expression defining correlation data. +When provided, this configures the underlying 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, 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 `amqp_publishConfirmNackCause` message header has been added. +It contains the `cause` of a 'nack' for publisher confirms. -<13> The channel to which returned messages are sent. +<14> Since _version 4.2_. The channel to which positive (ack) publisher confirms are sent; payload is the correlation data defined by the _confirm-correlation-expression_. +_Optional, default=nullChannel_. + +<15> Since _version 4.2_. The channel to which negative (nack) publisher confirms are sent; payload is the correlation data defined by the _confirm-correlation-expression_. +_Optional, default=nullChannel_. + +<16> 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 data received from amqp, with the following additional headers: _amqp_returnReplyCode, amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey_. _Optional_. -<14> When set to `false`, the endpoint will attempt to connect to the broker during application context initialization. +<17> When set to `false`, the endpoint will attempt to connect to the broker during application context initialization. This allows "fail fast" detection of bad configuration, but will also cause initialization to fail if the broker is down. When true (default), the connection is established (if it doesn't already exist because some other component established it) when the first message is sent. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index c6dc75a273..b83a7f2ef6 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -99,3 +99,11 @@ See <> for more information. Much more flexibility is now provided for dynamic polling. See <> for more information. + +[[x4.2-amqp-changes]] +==== AMQP Changes + +The `` now supports `confirm-correlation-expression` and `confirm-(n)ack-channel` +attributes with similar purpose as for ``. + +See <> for more information.