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

INT-3582: Pub Confirms for AMQP Outbound Gateway

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3582
This commit is contained in:
Artem Bilan
2015-04-28 16:12:58 +03:00
parent 6302c86fb8
commit 135287994b
8 changed files with 166 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -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<String>(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;
}

86
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd
View File

@@ -43,49 +43,6 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-correlation-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Expression for correlating publisher confirms to sent messages. The Rabbit API only correlates confirms
to a channel; this is used to further correlate a confirm to a message. An example is
"headers['amqp_confirmCorrelationData']", assuming that header contains the data.
Messages that do not have correlation data do not generate publisher confirm messages.
Requires a CachingConnectionFactory with the 'publisherConfirms' property set to TRUE.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-ack-channel" type="xsd:string" default="nullChannel">
<xsd:annotation>
<xsd:documentation><![CDATA[
Channel to which positive publisher confirms will be sent; the payload will be the correlation data from
the sent message. The message will also contain a header 'amqp_publishConfirm' with a value true.
Requires a connection factory that is configured to request publisher confirms. Default
is nullChannel in case an adapter doesn't want to use publisher confirms, but is using a
connection factory that is configured to request them.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-nack-channel" type="xsd:string" default="nullChannel">
<xsd:annotation>
<xsd:documentation><![CDATA[
Channel to which negative publisher confirms will be sent; the payload will be the correlation data from
the sent message. The message will also contain a header 'amqp_publishConfirm' with a value false.
Requires a connection factory that is configured to request publisher confirms. Default
is nullChannel in case an adapter doesn't want to use publisher confirms, but is using a
connection factory that is configured to request them.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -501,6 +458,49 @@ property set to TRUE.
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="confirm-correlation-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Expression for correlating publisher confirms to sent messages. The Rabbit API only correlates confirms
to a channel; this is used to further correlate a confirm to a message. An example is
"headers['amqp_confirmCorrelationData']", assuming that header contains the data.
Messages that do not have correlation data do not generate publisher confirm messages.
Requires a CachingConnectionFactory with the 'publisherConfirms' property set to TRUE.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-ack-channel" type="xsd:string" default="nullChannel">
<xsd:annotation>
<xsd:documentation><![CDATA[
Channel to which positive publisher confirms will be sent; the payload will be the correlation data from
the sent message. The message will also contain a header 'amqp_publishConfirm' with a value true.
Requires a connection factory that is configured to request publisher confirms. Default
is nullChannel in case an adapter doesn't want to use publisher confirms, but is using a
connection factory that is configured to request them.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-nack-channel" type="xsd:string" default="nullChannel">
<xsd:annotation>
<xsd:documentation><![CDATA[
Channel to which negative publisher confirms will be sent; the payload will be the correlation data from
the sent message. The message will also contain a header 'amqp_publishConfirm' with a value false.
Requires a connection factory that is configured to request publisher confirms. Default
is nullChannel in case an adapter doesn't want to use publisher confirms, but is using a
connection factory that is configured to request them.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:simpleType name="deliveryModeEnumeration">

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

@@ -80,4 +80,21 @@
mapped-reply-headers=""/>
</int:chain>
<rabbit:template id="amqpTemplateConfirms" connection-factory="connectionFactory" reply-timeout="10"/>
<amqp:outbound-gateway request-channel="pcRequestChannel"
reply-channel="fromRabbit"
exchange-name="si.test.exchange"
mapped-request-headers="foo*"
requires-reply="false"
amqp-template="amqpTemplateConfirms"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel"/>
<int:channel id="pcRequestChannel"/>
<int:channel id="ackChannel">
<int:queue/>
</int:channel>
</beans>

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

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

10
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java
View File

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

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

@@ -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.

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

@@ -99,3 +99,11 @@ See <<jms-message-driven-channel-adapter>> for more information.
Much more flexibility is now provided for dynamic polling.
See <<conditional-pollers>> for more information.
[[x4.2-amqp-changes]]
==== AMQP Changes
The `<int-amqp:outbound-gateway>` now supports `confirm-correlation-expression` and `confirm-(n)ack-channel`
attributes with similar purpose as for `<int-amqp:outbound-channel-adapter>`.
See <<amqp>> for more information.