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

INT-2394 Support HA (Mirrored) Queues

Browse Source 
Changes to Connection Factory interface (stub).

INT-2394 Polishing

INT-2394 Map Gateway Correlation Header

Map spring_reply_correlation if present.

INT-2394 Map ReplyTo Stack

Enable the use outbound-gateway in a destination application
to which we sent a message using another outbound gateway.

The RabbitTemplate now pushed exising correlation data onto
a stacked header attribute.

INT-2509 Publisher Confirms and Returns

Add Publisher Confirm support to the outbound channel
adapter.

Add returns support to outbound channel adapter and gateway.

INT-2509 Polishing

PR Review comments.

Docs.

Fix chain test after rebase.
This commit is contained in:
Gary Russell
2012-01-09 16:10:30 -05:00
committed by Oleg Zhurakousky
parent 8fbc9264f7
commit 11e5fe0c0c
16 changed files with 667 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -47,7 +47,7 @@ subprojects { subproject ->
mockitoVersion = '1.9.0'
// When changing Spring Versions - don't forget to update bundlor ranges
springVersion = '3.1.1.RELEASE'
springAmqpVersion = '1.0.0.RELEASE'
springAmqpVersion = '1.1.0.RELEASE'
springDataMongoVersion = '1.1.0.M1'
springDataRedisVersion = '1.0.0.RELEASE'
springGemfireVersion = '1.1.0.RELEASE'

14
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java
View File

@@ -72,4 +72,18 @@ public abstract class AmqpHeaders {
public static final String USER_ID = PREFIX + "userId";
public static final String SPRING_REPLY_CORRELATION = PREFIX + "springReplyCorrelation";
public static final String SPRING_REPLY_TO_STACK = PREFIX + "springReplyToStack";
public static final String PUBLISH_CONFIRM = PREFIX + "publishConfirm";
public static final String RETURN_REPLY_CODE = PREFIX + "returnReplyCode";
public static final String RETURN_REPLY_TEXT = PREFIX + "returnReplyText";
public static final String RETURN_EXCHANGE = PREFIX + "returnExchange";
public static final String RETURN_ROUTING_KEY = PREFIX + "returnRoutingKey";
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2012 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.
@@ -31,6 +31,7 @@ import org.w3c.dom.Element;
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @since 2.1
*/
public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@@ -47,9 +48,14 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-correlation-expression");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
return builder.getBeanDefinition();
}

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

@@ -51,7 +51,8 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
return builder;

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

@@ -20,12 +20,16 @@ import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.SpelParserConfiguration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -41,7 +45,8 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
implements RabbitTemplate.ConfirmCallback, ReturnCallback {
private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
@@ -64,11 +69,23 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
private volatile String confirmCorrelationExpression;
private volatile ExpressionEvaluatingMessageProcessor<Object> correlationDataGenerator;
private volatile MessageChannel confirmAckChannel;
private volatile MessageChannel confirmNackChannel;
private volatile MessageChannel returnChannel;
@Override
protected void onInit() {
super.onInit();
Assert.state(exchangeNameExpression == null || "".equals(exchangeName),
"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
Assert.state(this.confirmCorrelationExpression != null ? !this.expectReply : true,
"Confirm correlation expression does not apply to a gateway");
if (exchangeNameExpression != null) {
Expression expression = expressionParser.parseExpression(this.exchangeNameExpression);
this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
@@ -79,6 +96,16 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
Expression expression = expressionParser.parseExpression(this.routingKeyExpression);
this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
}
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 (this.returnChannel != null) {
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for publisher returns");
((RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
}
}
public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
@@ -110,6 +137,22 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
this.expectReply = expectReply;
}
public void setConfirmCorrelationExpression(String confirmCorrelationExpression) {
this.confirmCorrelationExpression = confirmCorrelationExpression;
}
public void setConfirmAckChannel(MessageChannel ackChannel) {
this.confirmAckChannel = ackChannel;
}
public void setConfirmNackChannel(MessageChannel nackChannel) {
this.confirmNackChannel = nackChannel;
}
public void setReturnChannel(MessageChannel returnChannel) {
this.returnChannel = returnChannel;
}
@Override
public String getComponentType() {
return expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter";
@@ -119,6 +162,19 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
protected Object handleRequestMessage(Message<?> requestMessage) {
String exchangeName = this.exchangeName;
String routingKey = this.routingKey;
CorrelationData correlationData = null;
if (this.correlationDataGenerator != null) {
Object userCorrelationData = this.correlationDataGenerator
.processMessage(requestMessage);
if (userCorrelationData != null) {
if (userCorrelationData instanceof CorrelationData) {
correlationData = (CorrelationData) userCorrelationData;
} else {
correlationData = new CorrelationDataWrapper(requestMessage
.getHeaders().getId().toString(), userCorrelationData);
}
}
}
if (this.exchangeNameGenerator != null) {
exchangeName = this.exchangeNameGenerator.processMessage(requestMessage);
}
@@ -129,20 +185,34 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
return this.sendAndReceive(exchangeName, routingKey, requestMessage);
}
else {
this.send(exchangeName, routingKey, requestMessage);
this.send(exchangeName, routingKey, requestMessage, correlationData);
return null;
}
}
private void send(String exchangeName, String routingKey, final Message<?> requestMessage) {
this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
new MessagePostProcessor() {
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
return message;
}
});
private void send(String exchangeName, String routingKey,
final Message<?> requestMessage, CorrelationData correlationData) {
if (this.amqpTemplate instanceof RabbitTemplate) {
((RabbitTemplate) this.amqpTemplate).convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
new MessagePostProcessor() {
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
return message;
}
},
correlationData);
}
else {
this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
new MessagePostProcessor() {
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
return message;
}
});
}
}
private Message<?> sendAndReceive(String exchangeName, String routingKey, Message<?> requestMessage) {
@@ -165,4 +235,58 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
return builder.build();
}
public void confirm(CorrelationData correlationData, boolean ack) {
Object userCorrelationData = correlationData;
if (correlationData instanceof CorrelationDataWrapper) {
userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData();
}
Message<Object> confirmMessage = MessageBuilder.withPayload(userCorrelationData)
.setHeader(AmqpHeaders.PUBLISH_CONFIRM, ack)
.build();
if (ack && this.confirmAckChannel != null) {
this.confirmAckChannel.send(confirmMessage);
}
else if (!ack && this.confirmNackChannel != null) {
this.confirmNackChannel.send(confirmMessage);
}
else {
if (logger.isInfoEnabled()) {
logger.info("Nowhere to send publisher confirm "
+ (ack ? "ack" : "nack") + " for "
+ userCorrelationData);
}
}
}
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;
}
}
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 onInit()
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
Object returnedObject = converter.fromMessage(message);
MessageBuilder<?> builder = (returnedObject instanceof Message)
? MessageBuilder.fromMessage((Message<?>) returnedObject)
: MessageBuilder.withPayload(returnedObject);
Map<String, ?> headers = this.headerMapper.toHeadersFromReply(message.getMessageProperties());
builder.copyHeadersIfAbsent(headers)
.setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)
.setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText)
.setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange)
.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey);
this.returnChannel.send(builder.build());
}
}

28
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
View File

@@ -24,6 +24,7 @@ import java.util.Map;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.mapping.AbstractHeaderMapper;
@@ -70,6 +71,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
STANDARD_HEADER_NAMES.add(AmqpHeaders.TIMESTAMP);
STANDARD_HEADER_NAMES.add(AmqpHeaders.TYPE);
STANDARD_HEADER_NAMES.add(AmqpHeaders.USER_ID);
STANDARD_HEADER_NAMES.add(AmqpHeaders.SPRING_REPLY_CORRELATION);
STANDARD_HEADER_NAMES.add(AmqpHeaders.SPRING_REPLY_TO_STACK);
}
/**
@@ -155,6 +158,18 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
if (StringUtils.hasText(userId)) {
headers.put(AmqpHeaders.USER_ID, userId);
}
Object replyCorrelation = amqpMessageProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER);
if (replyCorrelation instanceof String) {
if (StringUtils.hasText((String) replyCorrelation)) {
headers.put(AmqpHeaders.SPRING_REPLY_CORRELATION, replyCorrelation);
}
}
Object replyToStack = amqpMessageProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER);
if (replyToStack instanceof String) {
if (StringUtils.hasText((String) replyToStack)) {
headers.put(AmqpHeaders.SPRING_REPLY_TO_STACK, replyToStack);
}
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
@@ -169,7 +184,10 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
*/
@Override
protected Map<String, Object> extractUserDefinedHeaders(MessageProperties amqpMessageProperties) {
return amqpMessageProperties.getHeaders();
Map<String, Object> headers = amqpMessageProperties.getHeaders();
headers.remove(RabbitTemplate.STACKED_CORRELATION_HEADER);
headers.remove(RabbitTemplate.STACKED_REPLY_TO_HEADER);
return headers;
}
/**
@@ -254,6 +272,14 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
if (StringUtils.hasText(userId)) {
amqpMessageProperties.setUserId(userId);
}
String replyCorrelation = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class);
if (StringUtils.hasLength(replyCorrelation)) {
amqpMessageProperties.setHeader("spring_reply_correlation", replyCorrelation);
}
String replyToStack = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_TO_STACK, String.class);
if (StringUtils.hasLength(replyToStack)) {
amqpMessageProperties.setHeader("spring_reply_to", replyToStack);
}
}
@Override

402
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.2.xsd
View File

@@ -18,91 +18,75 @@
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Unique ID for this adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
If this attribute is not provided, the ID will be used to create a new DirectChannel, and then instead of using that
ID as the bean name of the EventDrivenConsumer instance that hosts the MessageHandler responsible for publishing the
AMQP Messages, that EventDrivenConsumer's bean name will be the ID plus the added suffix: ".adapter"
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exchange-name" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The fixed name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exchange-name-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="routing-key" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The fixed routing-key to use when sending Messages. By default, this will be an empty String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="routing-key-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The routing-key to use when sending Messages evaluated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="amqp-template" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:complexContent>
<xsd:extension base="outboundType">
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Unique ID for this adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
If this attribute is not provided, the ID will be used to create a new DirectChannel, and then instead of using that
ID as the bean name of the EventDrivenConsumer instance that hosts the MessageHandler responsible for publishing the
AMQP Messages, that EventDrivenConsumer's bean name will be the ID plus the added suffix: ".adapter"
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel" />
</tool:annotation>
</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.
]]></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.integration.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.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
@@ -142,112 +126,54 @@ this list can also be simple patterns to be matched against the header names (e.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Unique ID for this gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which replies should be sent after being received from an AQMP Queue and converted.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exchange-name" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exchange-name-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="routing-key" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The routing-key to use when sending Messages. By default, this will be an empty String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="routing-key-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The routing-key to use when sending Messages evealuated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="amqp-template" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-reply-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
which itself is a Map.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:complexContent>
<xsd:extension base="outboundType">
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Unique ID for this gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which replies should be sent after being received from an AQMP Queue and converted.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-reply-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
which itself is a Map.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
@@ -287,9 +213,9 @@ this list can also be simple patterns to be matched against the header names (e.
</xsd:attribute>
<xsd:attribute name="mapped-reply-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
which itself is a Map.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
@@ -365,7 +291,7 @@ this list can also be simple patterns to be matched against the header names (e.
Reference to a FanoutExchange instance to which this channel should send Messages. If not provided,
a FanoutExchange will be declared with this channel's name prefixed by "si.fanout.".
A Queue will be declared automatically and bound to that exchange to handle the consumer role
of this channel.
of this channel.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -412,6 +338,88 @@ this list can also be simple patterns to be matched against the header names (e.
<xsd:attributeGroup ref="containerAndTemplateAttributes"/>
</xsd:complexType>
<xsd:complexType name="outboundType">
<xsd:annotation>
<xsd:documentation>
Base type for the 'outbound-channel-adapter' and 'outbound-gateway' elements.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="exchange-name" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The fixed name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exchange-name-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="routing-key" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The fixed routing-key to use when sending Messages. By default, this will be an empty String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="routing-key-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The routing-key to use when sending Messages evaluated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="amqp-template" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="return-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Channel to which returned messages will be sent.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="inboundType">
<xsd:annotation>
<xsd:documentation>
@@ -527,7 +535,7 @@ this list can also be simple patterns to be matched against the header names (e.
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to the Rabbit ConnectionFactory to be used by this component.
Reference to the Rabbit ConnectionFactory to be used by this component.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -583,7 +591,7 @@ this list can also be simple patterns to be matched against the header names (e.
<xsd:attributeGroup name="containerOnlyAttributes">
<xsd:annotation>
<xsd:documentation>
Attributes for a SimpleMesssageListenerContainer's properties other than queues, queueNames, messageListener, and
Attributes for a SimpleMesssageListenerContainer's properties other than queues, queueNames, messageListener, and
autoStartup which may or may not be exposed for configuration depending on what type of component uses this attribute group.
This group also does not include any of the properties that are shared with RabbitTemplate, such as channelTransacted,
connectionFactory, and messsagePropertiesConverter.
@@ -692,7 +700,7 @@ this list can also be simple patterns to be matched against the header names (e.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

56
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.amqp;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.connection.Connection;
@@ -25,6 +26,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Command;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.FlowListener;
@@ -47,6 +49,7 @@ import com.rabbitmq.client.AMQP.Tx.RollbackOk;
/**
* @author Mark Fisher
* @author Gary Russell
* @since 2.1
*/
public class StubRabbitConnectionFactory implements ConnectionFactory {
@@ -132,24 +135,29 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
public void abort(int closeCode, String closeMessage) throws IOException {
}
@SuppressWarnings("unused")
public ReturnListener getReturnListener() {
return null;
}
public void setReturnListener(ReturnListener listener) {
public void addReturnListener(ReturnListener listener) {
}
@SuppressWarnings("unused")
public FlowListener getFlowListener() {
return null;
}
@SuppressWarnings("unused")
public void setFlowListener(FlowListener listener) {
}
@SuppressWarnings("unused")
public ConfirmListener getConfirmListener() {
return null;
}
@SuppressWarnings("unused")
public void setConfirmListener(ConfirmListener listener) {
}
@@ -354,9 +362,53 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
public void asyncRpc(Method method) throws IOException {
}
public Method rpc(Method method) throws IOException {
public Command rpc(Method method) throws IOException {
return null;
}
public boolean removeReturnListener(ReturnListener listener) {
return false;
}
public void clearReturnListeners() {
}
public void addFlowListener(FlowListener listener) {
}
public boolean removeFlowListener(FlowListener listener) {
return false;
}
public void clearFlowListeners() {
}
public void addConfirmListener(ConfirmListener listener) {
}
public boolean removeConfirmListener(ConfirmListener listener) {
return false;
}
public void clearConfirmListeners() {
}
public boolean waitForConfirms() throws InterruptedException {
return false;
}
public void waitForConfirmsOrDie() throws IOException,
InterruptedException {
}
public boolean waitForConfirms(long timeout)
throws InterruptedException, TimeoutException {
return false;
}
public void waitForConfirmsOrDie(long timeout) throws IOException,
InterruptedException, TimeoutException {
}
}
}

7
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java
View File

@@ -28,6 +28,7 @@ import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
@@ -105,7 +106,8 @@ public class AmqpInboundGatewayParserTests {
assertEquals("bar", properties.getHeaders().get("bar"));
return null;
}})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class));
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(Message.class), Mockito.any(CorrelationData.class));
ReflectionUtils.setField(amqpTemplateField, gateway, amqpTemplate);
AbstractMessageListenerContainer mlc =
@@ -123,7 +125,8 @@ public class AmqpInboundGatewayParserTests {
Message amqpMessage = new Message("hello".getBytes(), amqpProperties);
listener.onMessage(amqpMessage);
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class));
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(Message.class), Mockito.any(CorrelationData.class));
}
private static class TestConverter extends SimpleMessageConverter {}

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

@@ -16,16 +16,38 @@
<bean id="connectionFactory" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
</bean>
<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel"
<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"/>
<int:channel id="requestChannel"/>
<int:channel id="requestChannel"/>
<int:chain id="chainWithRabbitOutbound" input-channel="amqpOutboundChannelAdapterWithinChain">
<amqp:outbound-channel-adapter exchange-name="outboundchanneladapter.test.1"/>
</int:chain>
<amqp:outbound-channel-adapter id="withPublisherConfirms" channel="pcRequestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel"/>
<int:channel id="pcRequestChannel"/>
<int:channel id="ackChannel">
<int:queue/>
</int:channel>
<amqp:outbound-channel-adapter id="withReturns" channel="returnRequestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"
return-channel="returnChannel"/>
<int:channel id="returnRequestChannel"/>
<int:channel id="returnChannel">
<int:queue/>
</int:channel>
</beans>

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

@@ -16,8 +16,17 @@
package org.springframework.integration.amqp.config;
import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.lang.reflect.Field;
import java.util.List;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -25,14 +34,22 @@ import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
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.PublisherCallbackChannel;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.support.MessageBuilder;
@@ -41,9 +58,8 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
import static org.junit.Assert.assertNull;
import static junit.framework.Assert.assertEquals;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
/**
* @author Mark Fisher
@@ -70,36 +86,70 @@ public class AmqpOutboundChannelAdapterParserTests {
assertEquals("amqp:outbound-channel-adapter", ((AmqpOutboundEndpoint) handler).getComponentType());
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperCustomHeaders() {
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomHeaders");
Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomHeaders");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
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();
assertEquals("foo", properties.getHeaders().get("foo"));
assertEquals("foobar", properties.getHeaders().get("foobar"));
assertNull(properties.getHeaders().get("bar"));
return null;
}})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Mockito.doAnswer(new Answer<Object>() {
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();
assertEquals("foo", properties.getHeaders().get("foo"));
assertEquals("foobar", properties.getHeaders().get("foobar"));
assertNull(properties.getHeaders().get("bar"));
return null;
}})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("requestChannel", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").setHeader("bar", "bar").setHeader("foobar", "foobar").build();
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.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));
}
@Test
public void parseWithPublisherConfirms() {
Object eventDrivenConsumer = context.getBean("withPublisherConfirms");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
NullChannel nullChannel = context.getBean(NullChannel.class);
MessageChannel ackChannel = context.getBean("ackChannel", MessageChannel.class);
assertSame(ackChannel, TestUtils.getPropertyValue(endpoint, "confirmAckChannel"));
assertSame(nullChannel, TestUtils.getPropertyValue(endpoint, "confirmNackChannel"));
}
@Test
public void withPublisherConfirms() throws Exception {
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(mockConnection);
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
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));
}
@SuppressWarnings("rawtypes")
@@ -123,14 +173,45 @@ public class AmqpOutboundChannelAdapterParserTests {
assertEquals("hello", new String(amqpReplyMessage.getBody()));
return null;
}})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class),
Mockito.any(CorrelationData.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("amqpOutboundChannelAdapterWithinChain", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").build();
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.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));
}
@Test
public void withReturns() throws Exception {
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(mockConnection);
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
MessageChannel requestChannel = context.getBean("returnRequestChannel", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").build();
requestChannel.send(message);
PollableChannel returnChannel = context.getBean("returnChannel", PollableChannel.class);
RabbitTemplate template = context.getBean(RabbitTemplate.class);
Map<String, Object> headers = new HashMap<String, Object>();
headers.put(PublisherCallbackChannel.RETURN_CORRELATION, template.getUUID());
BasicProperties properties = mock(BasicProperties.class);
when(properties.getHeaders()).thenReturn(headers);
when(properties.getContentType()).thenReturn("text/plain");
publisherCallbackChannel.handleReturn(123, "reply text", "anExchange", "bar", properties, "hello".getBytes());
Message<?> returned = returnChannel.receive(1000);
assertNotNull(returned);
assertEquals(123, returned.getHeaders().get(AmqpHeaders.RETURN_REPLY_CODE));
assertEquals("reply text", returned.getHeaders().get(AmqpHeaders.RETURN_REPLY_TEXT));
assertEquals("anExchange", returned.getHeaders().get(AmqpHeaders.RETURN_EXCHANGE));
assertEquals("bar", returned.getHeaders().get(AmqpHeaders.RETURN_ROUTING_KEY));
assertEquals("hello", returned.getPayload());
}
}

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

@@ -14,7 +14,8 @@
exchange-name="si.test.exchange"
routing-key="si.test.binding"
amqp-template="amqpTemplate"
order="5"/>
order="5"
return-channel="returnChannel"/>
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>
@@ -54,4 +55,8 @@
mapped-request-headers=""
mapped-reply-headers=""/>
<int:channel id="returnChannel">
<int:queue/>
</int:channel>
</beans>

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

@@ -38,6 +38,7 @@ import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
/**
@@ -57,6 +58,8 @@ public class AmqpOutboundGatewayParserTests {
assertTrue(context.containsBean("rabbitGateway"));
assertEquals(context.getBean("fromRabbit"), TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals("amqp:outbound-gateway", gateway.getComponentType());
MessageChannel returnChannel = context.getBean("returnChannel", MessageChannel.class);
assertSame(returnChannel, TestUtils.getPropertyValue(gateway, "returnChannel"));
}
@SuppressWarnings("rawtypes")

13
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2012 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.
@@ -26,6 +26,7 @@ import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.JsonMessageConverter;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.amqp.AmqpHeaders;
@@ -35,6 +36,7 @@ import static org.junit.Assert.fail;
/**
* @author Mark Fisher
* @author Gary Russell
* @since 2.1
*/
public class DefaultAmqpHeaderMapperTests {
@@ -62,6 +64,8 @@ public class DefaultAmqpHeaderMapperTests {
headerMap.put(AmqpHeaders.TIMESTAMP, testTimestamp);
headerMap.put(AmqpHeaders.TYPE, "test.type");
headerMap.put(AmqpHeaders.USER_ID, "test.userId");
headerMap.put(AmqpHeaders.SPRING_REPLY_CORRELATION, "test.correlation");
headerMap.put(AmqpHeaders.SPRING_REPLY_TO_STACK, "test.replyTo2");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
@@ -88,6 +92,8 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals(testTimestamp, amqpProperties.getTimestamp());
assertEquals("test.type", amqpProperties.getType());
assertEquals("test.userId", amqpProperties.getUserId());
assertEquals("test.correlation", amqpProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER));
assertEquals("test.replyTo2", amqpProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER));
}
@Test
@@ -115,6 +121,8 @@ public class DefaultAmqpHeaderMapperTests {
amqpProperties.setTimestamp(testTimestamp);
amqpProperties.setType("test.type");
amqpProperties.setUserId("test.userId");
amqpProperties.setHeader(RabbitTemplate.STACKED_CORRELATION_HEADER, "test.correlation");
amqpProperties.setHeader(RabbitTemplate.STACKED_REPLY_TO_HEADER, "test.replyTo2");
Map<String, Object> headerMap = headerMapper.toHeadersFromReply(amqpProperties);
assertEquals("test.appId", headerMap.get(AmqpHeaders.APP_ID));
assertEquals("test.clusterId", headerMap.get(AmqpHeaders.CLUSTER_ID));
@@ -133,6 +141,8 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals(testTimestamp, headerMap.get(AmqpHeaders.TIMESTAMP));
assertEquals("test.type", headerMap.get(AmqpHeaders.TYPE));
assertEquals("test.userId", headerMap.get(AmqpHeaders.USER_ID));
assertEquals("test.correlation", headerMap.get(AmqpHeaders.SPRING_REPLY_CORRELATION));
assertEquals("test.replyTo2", headerMap.get(AmqpHeaders.SPRING_REPLY_TO_STACK));
}
@Test
@@ -171,4 +181,5 @@ public class DefaultAmqpHeaderMapperTests {
Object result = converter.fromMessage(new Message("123".getBytes(), amqpProperties));
assertEquals(String.class, result.getClass());
}
}

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

@@ -280,7 +280,11 @@ this list can also be simple patterns to be matched against the header names (e.
exchange-name=""]]><co id="amqp-outbound-channel-adapter-xml-4-co" linkends="amqp-outbound-channel-adapter-xml-4" /><![CDATA[
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[/>]]>
routing-key-expression=""]]><co id="amqp-outbound-channel-adapter-xml-7-co" linkends="amqp-outbound-channel-adapter-xml-7" /><![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[
return-channel=""]]><co id="amqp-outbound-channel-adapter-xml-11-co" linkends="amqp-outbound-channel-adapter-xml-11" /><![CDATA[/>]]>
</programlisting>
<para>
<calloutlist>
@@ -321,6 +325,33 @@ 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-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 RabbitTemplate and a
CachingConnectionFactory with publisherConfirms enabled. When a publisher confirm
is received, 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".
<emphasis>Optional</emphasis>.</para>
</callout>
<callout arearefs="amqp-outbound-channel-adapter-xml-9-co" id="amqp-outbound-channel-adapter-xml-9">
<para>The channel to which positive (ack) publisher confirms are sent; payload is
the correlation data defined by the <emphasis>confirm-correlation-expression</emphasis>.
<emphasis>Optional, default=nullChannel</emphasis>.</para>
</callout>
<callout arearefs="amqp-outbound-channel-adapter-xml-10-co" id="amqp-outbound-channel-adapter-xml-10">
<para>The channel to which negative (nack) publisher confirms are sent; payload is
the correlation data defined by the <emphasis>confirm-correlation-expression</emphasis>.
<emphasis>Optional, default=nullChannel</emphasis>.</para>
</callout>
<callout arearefs="amqp-outbound-channel-adapter-xml-11-co" id="amqp-outbound-channel-adapter-xml-11">
<para>The channel to which returned messages are sent. When provided, the underlying amqp template
is configured to return undeliverable messages. The message will be constructed from the
data received from amqp, with the following additional headers: <emphasis>amqp_returnReplyCode,
amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey</emphasis>.
<emphasis>Optional</emphasis>.</para>
</callout>
</calloutlist>
</para>
</section>
@@ -387,8 +418,9 @@ this list can also be simple patterns to be matched against the header names (e.
exchange-name=""]]><co id="amqp-outbound-gateway-adapter-xml-4-co" linkends="amqp-outbound-gateway-adapter-xml-4" /><![CDATA[
order="1"]]><co id="amqp-outbound-gateway-adapter-xml-5-co" linkends="amqp-outbound-gateway-adapter-xml-5" /><![CDATA[
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[/>]]>
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[
return-channel=""]]><co id="amqp-outbound-gateway-adapter-xml-9-co" linkends="amqp-outbound-gateway-adapter-xml-9" /><![CDATA[/>]]>
</programlisting>
<para>
<calloutlist>
@@ -434,8 +466,25 @@ 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>
</calloutlist>
</para>
<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. The message will be constructed from the
data received from amqp, with the following additional headers: <emphasis>amqp_returnReplyCode,
amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey</emphasis>.
<emphasis>Optional</emphasis>.</para>
</callout>
</calloutlist>
</para>
<note>
<para>Prior to Spring Integration 2.2, and Spring AMQP 1.1, the outbound gateway used a new, temporary,
reply queue for each request. This is still the default, but now the RabbitTemplate can be configured
with a specific queue for replies; headers are added to the outbound message for request/reply correlation.
It is important that the consuming application returns these headers unchanged. The headers are
<classname>spring_reply_correlation</classname> and <classname>spring_reply_to</classname>. If
the consuming application is a Spring Integration application, these headers will be managed
automatically, including the case where that application might send a request/reply to a third
application using an outbound gateway.</para>
</note>
</section>

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

@@ -18,6 +18,20 @@
Spring Integration now uses Spring 3.1.
</para>
</section>
<section id="2.2-amqp-11">
<title>Spring-AMQP 1.1</title>
<para>
Spring Integration now uses Spring AMQP 1.1. This enables several features
to be used within a Spring Integration application, including...
</para>
<itemizedlist>
<listitem>A fixed reply queue for the outbound gateway</listitem>
<listitem>HA (mirrored) queues</listitem>
<listitem>Publisher Confirms</listitem>
<listitem>Returned Messages</listitem>
<listitem>Support for Dead Letter Exchanges/Dead Letter Queues</listitem>
</itemizedlist>
</section>
</section>
<section id="2.2-new-components">