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

INT-3943: AMQP Async Outbound Gateway

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3943

Initial commit.

Polishing; Address PR Comments; Docs

Doc Polishing

Async GW - Support requiresReply

Polishing.

Avoid extra `requiresReply` in the `AsyncAmqpOutboundGateway`
This commit is contained in:
Gary Russell
2016-03-01 15:54:53 -05:00
committed by Artem Bilan
parent 78aaa6dac6
commit d4e6615f82
16 changed files with 1357 additions and 444 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2016 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.
@@ -20,12 +20,14 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
/**
* Namespace handler for the AMQP schema.
*
*
* @author Mark Fisher
* @author Gary Russell
* @since 2.1
*/
public class AmqpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@Override
public void init() {
this.registerBeanDefinitionParser("channel", new AmqpChannelParser());
this.registerBeanDefinitionParser("publish-subscribe-channel", new AmqpChannelParser());
@@ -33,6 +35,7 @@ public class AmqpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
this.registerBeanDefinitionParser("inbound-gateway", new AmqpInboundGatewayParser());
this.registerBeanDefinitionParser("outbound-channel-adapter", new AmqpOutboundChannelAdapterParser());
this.registerBeanDefinitionParser("outbound-gateway", new AmqpOutboundGatewayParser());
this.registerBeanDefinitionParser("outbound-async-gateway", new AmqpOutboundGatewayParser());
}
}

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

@@ -22,6 +22,7 @@ import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.outbound.AsyncAmqpOutboundGateway;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
@@ -47,13 +48,25 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AmqpOutboundEndpoint.class);
String amqpTemplateRef = element.getAttribute("amqp-template");
if (!StringUtils.hasText(amqpTemplateRef)) {
amqpTemplateRef = "amqpTemplate";
BeanDefinitionBuilder builder;
boolean async = element.getLocalName().contains("async");
if (async) {
builder = BeanDefinitionBuilder.genericBeanDefinition(AsyncAmqpOutboundGateway.class);
String asyncTemplateRef = element.getAttribute("async-template");
if (!StringUtils.hasText(asyncTemplateRef)) {
asyncTemplateRef = "asyncRabbitTemplate";
}
builder.addConstructorArgReference(asyncTemplateRef);
}
else {
builder = BeanDefinitionBuilder.genericBeanDefinition(AmqpOutboundEndpoint.class);
String amqpTemplateRef = element.getAttribute("amqp-template");
if (!StringUtils.hasText(amqpTemplateRef)) {
amqpTemplateRef = "amqpTemplate";
}
builder.addConstructorArgReference(amqpTemplateRef);
builder.addPropertyValue("expectReply", true);
}
builder.addConstructorArgReference(amqpTemplateRef);
builder.addPropertyValue("expectReply", true);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name", true);
BeanDefinition exchangeNameExpression =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("exchange-name-expression", element);

473
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java Normal file
View File

@@ -0,0 +1,473 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.amqp.outbound;
import java.util.HashMap;
import java.util.Map;
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.support.CorrelationData;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.expression.Expression;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Gary Russell
* @since 4.3
*
*/
public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
implements Lifecycle {
private volatile String exchangeName;
private volatile String routingKey;
private volatile Expression exchangeNameExpression;
private volatile Expression routingKeyExpression;
private volatile ExpressionEvaluatingMessageProcessor<String> routingKeyGenerator;
private volatile ExpressionEvaluatingMessageProcessor<String> exchangeNameGenerator;
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
private volatile Expression confirmCorrelationExpression;
private volatile ExpressionEvaluatingMessageProcessor<Object> correlationDataGenerator;
private volatile MessageChannel confirmAckChannel;
private volatile MessageChannel confirmNackChannel;
private volatile MessageChannel returnChannel;
private volatile MessageDeliveryMode defaultDeliveryMode;
private volatile boolean lazyConnect = true;
private volatile ConnectionFactory connectionFactory;
private volatile boolean running;
public void setHeaderMapper(AmqpHeaderMapper headerMapper) {
Assert.notNull(headerMapper, "headerMapper must not be null");
this.headerMapper = headerMapper;
}
public void setExchangeName(String exchangeName) {
Assert.notNull(exchangeName, "exchangeName must not be null");
this.exchangeName = exchangeName;
}
/**
* @param exchangeNameExpression the expression to use.
* @since 4.3
*/
public void setExchangeNameExpression(Expression exchangeNameExpression) {
this.exchangeNameExpression = exchangeNameExpression;
}
/**
* @param exchangeNameExpression the String in SpEL syntax.
* @since 4.3
*/
public void setExchangeNameExpressionString(String exchangeNameExpression) {
Assert.hasText(exchangeNameExpression, "'exchangeNameExpression' must not be empty");
this.exchangeNameExpression = EXPRESSION_PARSER.parseExpression(exchangeNameExpression);
}
public void setRoutingKey(String routingKey) {
Assert.notNull(routingKey, "routingKey must not be null");
this.routingKey = routingKey;
}
/**
* @param routingKeyExpression the expression to use.
* @since 4.3
*/
public void setRoutingKeyExpression(Expression routingKeyExpression) {
this.routingKeyExpression = routingKeyExpression;
}
/**
* @param routingKeyExpression the String in SpEL syntax.
* @since 4.3
*/
public void setRoutingKeyExpressionString(String routingKeyExpression) {
Assert.hasText(routingKeyExpression, "'routingKeyExpression' must not be empty");
this.routingKeyExpression = EXPRESSION_PARSER.parseExpression(routingKeyExpression);
}
/**
* @param confirmCorrelationExpression the expression to use.
* @since 4.3
*/
public void setConfirmCorrelationExpression(Expression confirmCorrelationExpression) {
this.confirmCorrelationExpression = confirmCorrelationExpression;
}
/**
* @param confirmCorrelationExpression the String in SpEL syntax.
* @since 4.3
*/
public void setConfirmCorrelationExpressionString(String confirmCorrelationExpression) {
Assert.hasText(confirmCorrelationExpression, "'confirmCorrelationExpression' must not be empty");
this.confirmCorrelationExpression = EXPRESSION_PARSER.parseExpression(confirmCorrelationExpression);
}
/**
* Set the channel to which acks are send (publisher confirms).
* @param ackChannel the channel.
*/
public void setConfirmAckChannel(MessageChannel ackChannel) {
this.confirmAckChannel = ackChannel;
}
/**
* Set the channel to which nacks are send (publisher confirms).
* @param nackChannel the channel.
*/
public void setConfirmNackChannel(MessageChannel nackChannel) {
this.confirmNackChannel = nackChannel;
}
/**
* Set the channel to which returned messages are sent.
* @param returnChannel the channel.
*/
public void setReturnChannel(MessageChannel returnChannel) {
this.returnChannel = returnChannel;
}
/**
* Set the default delivery mode.
* @param defaultDeliveryMode the delivery mode.
*/
public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
this.defaultDeliveryMode = defaultDeliveryMode;
}
/**
* Set to {@code false} to attempt to connect during endpoint start;
* default {@code true}, meaning the connection will be attempted
* to be established on the arrival of the first message.
* @param lazyConnect the lazyConnect to set
* @since 4.1
*/
public void setLazyConnect(boolean lazyConnect) {
this.lazyConnect = lazyConnect;
}
protected final void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
protected String getExchangeName() {
return this.exchangeName;
}
protected String getRoutingKey() {
return this.routingKey;
}
protected Expression getExchangeNameExpression() {
return this.exchangeNameExpression;
}
protected Expression getRoutingKeyExpression() {
return this.routingKeyExpression;
}
protected ExpressionEvaluatingMessageProcessor<String> getRoutingKeyGenerator() {
return this.routingKeyGenerator;
}
protected ExpressionEvaluatingMessageProcessor<String> getExchangeNameGenerator() {
return this.exchangeNameGenerator;
}
protected AmqpHeaderMapper getHeaderMapper() {
return this.headerMapper;
}
protected Expression getConfirmCorrelationExpression() {
return this.confirmCorrelationExpression;
}
protected ExpressionEvaluatingMessageProcessor<Object> getCorrelationDataGenerator() {
return this.correlationDataGenerator;
}
protected MessageChannel getConfirmAckChannel() {
return this.confirmAckChannel;
}
protected MessageChannel getConfirmNackChannel() {
return this.confirmNackChannel;
}
protected MessageChannel getReturnChannel() {
return this.returnChannel;
}
protected MessageDeliveryMode getDefaultDeliveryMode() {
return this.defaultDeliveryMode;
}
protected boolean isLazyConnect() {
return this.lazyConnect;
}
@Override
protected final void doInit() {
Assert.state(this.exchangeNameExpression == null || this.exchangeName == null,
"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
BeanFactory beanFactory = getBeanFactory();
if (this.exchangeNameExpression != null) {
this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(this.exchangeNameExpression,
String.class);
if (beanFactory != null) {
this.exchangeNameGenerator.setBeanFactory(beanFactory);
}
}
Assert.state(routingKeyExpression == null || routingKey == null,
"Either a routingKey or a routingKeyExpression can be provided, but not both");
if (this.routingKeyExpression != null) {
this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor<String>(this.routingKeyExpression,
String.class);
if (beanFactory != null) {
this.routingKeyGenerator.setBeanFactory(beanFactory);
}
}
if (this.confirmCorrelationExpression != null) {
this.correlationDataGenerator =
new ExpressionEvaluatingMessageProcessor<Object>(this.confirmCorrelationExpression, Object.class);
if (beanFactory != null) {
this.correlationDataGenerator.setBeanFactory(beanFactory);
}
}
else {
NullChannel nullChannel = extractTypeIfPossible(this.confirmAckChannel, NullChannel.class);
Assert.state(this.confirmAckChannel == null || nullChannel != null,
"A 'confirmCorrelationExpression' is required when specifying a 'confirmAckChannel'");
nullChannel = extractTypeIfPossible(this.confirmNackChannel, NullChannel.class);
Assert.state(this.confirmNackChannel == null || nullChannel != null,
"A 'confirmCorrelationExpression' is required when specifying a 'confirmNackChannel'");
}
endpointInit();
}
/**
* Subclasses can override to perform any additional initialization.
* Called from afterPropertiesSet().
*/
protected void endpointInit() {
}
@Override
public synchronized void start() {
if (!this.running) {
if (!this.lazyConnect && this.connectionFactory != null) {
try {
Connection connection = connectionFactory.createConnection();
if (connection != null) {
connection.close();
}
}
catch (RuntimeException e) {
logger.error("Failed to eagerly establish the connection.", e);
}
}
doStart();
this.running = true;
}
}
@Override
public synchronized void stop() {
if (this.running) {
doStop();
}
this.running = false;
}
protected void doStart() {
}
protected void doStop() {
}
@Override
public boolean isRunning() {
return this.running;
}
protected CorrelationData generateCorrelationData(Message<?> requestMessage) {
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);
}
}
}
return correlationData;
}
protected String generateExchangeName(Message<?> requestMessage) {
String exchangeName = this.exchangeName;
if (this.exchangeNameGenerator != null) {
exchangeName = this.exchangeNameGenerator.processMessage(requestMessage);
}
return exchangeName;
}
protected String generateRoutingKey(Message<?> requestMessage) {
String routingKey = this.routingKey;
if (this.routingKeyGenerator != null) {
routingKey = this.routingKeyGenerator.processMessage(requestMessage);
}
return routingKey;
}
protected org.springframework.amqp.core.Message mapMessage(Message<?> requestMessage, MessageConverter converter) {
MessageProperties amqpMessageProperties = new MessageProperties();
org.springframework.amqp.core.Message amqpMessage;
if (converter instanceof ContentTypeDelegatingMessageConverter) {
getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
}
else { // See INT-3002 - map headers last if we're not using a CTDMC
amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
}
checkDeliveryMode(requestMessage, amqpMessageProperties);
return amqpMessage;
}
private void checkDeliveryMode(Message<?> requestMessage, MessageProperties messageProperties) {
if (this.defaultDeliveryMode != null &&
requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) {
messageProperties.setDeliveryMode(this.defaultDeliveryMode);
}
}
protected Message<?> buildReplyMessage(MessageConverter converter,
org.springframework.amqp.core.Message amqpReplyMessage) {
Object replyObject = converter.fromMessage(amqpReplyMessage);
AbstractIntegrationMessageBuilder<?> builder = (replyObject instanceof Message)
? this.getMessageBuilderFactory().fromMessage((Message<?>) replyObject)
: this.getMessageBuilderFactory().withPayload(replyObject);
Map<String, ?> headers = getHeaderMapper().toHeadersFromReply(amqpReplyMessage.getMessageProperties());
builder.copyHeadersIfAbsent(headers);
return builder.build();
}
protected Message<?> buildReturnedMessage(org.springframework.amqp.core.Message message,
int replyCode, String replyText, String exchange, String routingKey, MessageConverter converter) {
Object returnedObject = converter.fromMessage(message);
AbstractIntegrationMessageBuilder<?> builder = (returnedObject instanceof Message)
? this.getMessageBuilderFactory().fromMessage((Message<?>) returnedObject)
: this.getMessageBuilderFactory().withPayload(returnedObject);
Map<String, ?> headers = getHeaderMapper().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);
return builder.build();
}
protected void handleConfirm(CorrelationData correlationData, boolean ack, String cause) {
Object userCorrelationData = correlationData;
if (correlationData == null) {
if (logger.isDebugEnabled()) {
logger.debug("No correlation data provided for ack: " + ack + " cause:" + cause);
}
return;
}
if (correlationData instanceof CorrelationDataWrapper) {
userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData();
}
Map<String, Object> headers = new HashMap<String, Object>();
headers.put(AmqpHeaders.PUBLISH_CONFIRM, ack);
if (!ack && StringUtils.hasText(cause)) {
headers.put(AmqpHeaders.PUBLISH_CONFIRM_NACK_CAUSE, cause);
}
AbstractIntegrationMessageBuilder<?> builder = userCorrelationData instanceof Message
? this.getMessageBuilderFactory().fromMessage((Message<?>) userCorrelationData)
: this.getMessageBuilderFactory().withPayload(userCorrelationData);
Message<?> confirmMessage = builder
.copyHeaders(headers)
.build();
if (ack && confirmAckChannel != null) {
sendOutput(confirmMessage, this.confirmAckChannel, true);
}
else if (!ack && this.confirmNackChannel != null) {
sendOutput(confirmMessage, this.confirmNackChannel, true);
}
else {
if (logger.isInfoEnabled()) {
logger.info("Nowhere to send publisher confirm "
+ (ack ? "ack" : "nack") + " for "
+ userCorrelationData);
}
}
}
protected static class CorrelationDataWrapper extends CorrelationData {
private final Object userData;
private CorrelationDataWrapper(String id, Object userData) {
super(id);
this.userData = userData;
}
public Object getUserData() {
return this.userData;
}
}
}

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

@@ -16,37 +16,17 @@
package org.springframework.integration.amqp.outbound;
import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
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.AmqpHeaders;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.expression.Expression;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Adapter that converts and sends Messages to an AMQP Exchange.
@@ -57,72 +37,19 @@ import org.springframework.util.StringUtils;
* @author Artem Bilan
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
implements RabbitTemplate.ConfirmCallback, ReturnCallback,
ApplicationListener<ContextRefreshedEvent>, Lifecycle {
public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
implements RabbitTemplate.ConfirmCallback, ReturnCallback {
private final AmqpTemplate amqpTemplate;
private volatile boolean expectReply;
private volatile String exchangeName;
private volatile String routingKey;
private volatile Expression exchangeNameExpression;
private volatile Expression routingKeyExpression;
private volatile ExpressionEvaluatingMessageProcessor<String> routingKeyGenerator;
private volatile ExpressionEvaluatingMessageProcessor<String> exchangeNameGenerator;
private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper();
private volatile Expression confirmCorrelationExpression;
private volatile ExpressionEvaluatingMessageProcessor<Object> correlationDataGenerator;
private volatile MessageChannel confirmAckChannel;
private volatile MessageChannel confirmNackChannel;
private volatile MessageChannel returnChannel;
private volatile MessageDeliveryMode defaultDeliveryMode;
private volatile boolean lazyConnect = true;
public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
Assert.notNull(amqpTemplate, "amqpTemplate must not be null");
this.amqpTemplate = amqpTemplate;
}
public void setHeaderMapper(AmqpHeaderMapper headerMapper) {
Assert.notNull(headerMapper, "headerMapper must not be null");
this.headerMapper = headerMapper;
}
public void setExchangeName(String exchangeName) {
Assert.notNull(exchangeName, "exchangeName must not be null");
this.exchangeName = exchangeName;
}
/**
* @param exchangeNameExpression the expression to use.
* @since 4.3
*/
public void setExchangeNameExpression(Expression exchangeNameExpression) {
this.exchangeNameExpression = exchangeNameExpression;
}
/**
* @param exchangeNameExpression the String in SpEL syntax.
* @since 4.3
*/
public void setExchangeNameExpressionString(String exchangeNameExpression) {
Assert.hasText(exchangeNameExpression, "'exchangeNameExpression' must not be empty");
this.exchangeNameExpression = EXPRESSION_PARSER.parseExpression(exchangeNameExpression);
if (amqpTemplate instanceof RabbitTemplate) {
setConnectionFactory(((RabbitTemplate) amqpTemplate).getConnectionFactory());
}
}
/**
@@ -134,28 +61,6 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
setExchangeNameExpression(exchangeNameExpression);
}
public void setRoutingKey(String routingKey) {
Assert.notNull(routingKey, "routingKey must not be null");
this.routingKey = routingKey;
}
/**
* @param routingKeyExpression the expression to use.
* @since 4.3
*/
public void setRoutingKeyExpression(Expression routingKeyExpression) {
this.routingKeyExpression = routingKeyExpression;
}
/**
* @param routingKeyExpression the String in SpEL syntax.
* @since 4.3
*/
public void setRoutingKeyExpressionString(String routingKeyExpression) {
Assert.hasText(routingKeyExpression, "'routingKeyExpression' must not be empty");
this.routingKeyExpression = EXPRESSION_PARSER.parseExpression(routingKeyExpression);
}
/**
* @param routingKeyExpression the expression to set.
* @deprecated in favor of {@link #setRoutingKeyExpression}.
@@ -169,57 +74,13 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
this.expectReply = expectReply;
}
/**
* @param confirmCorrelationExpression the expression to use.
* @since 4.3
*/
public void setConfirmCorrelationExpression(Expression confirmCorrelationExpression) {
this.confirmCorrelationExpression = confirmCorrelationExpression;
}
/**
* @param confirmCorrelationExpression the String in SpEL syntax.
* @since 4.3
*/
public void setConfirmCorrelationExpressionString(String confirmCorrelationExpression) {
Assert.hasText(confirmCorrelationExpression, "'confirmCorrelationExpression' must not be empty");
this.confirmCorrelationExpression = EXPRESSION_PARSER.parseExpression(confirmCorrelationExpression);
}
/**
* @param confirmCorrelationExpression the expression to set.
* @deprecated in favor of {@link #setConfirmCorrelationExpression}.
*/
@Deprecated
public void setExpressionConfirmCorrelation(Expression confirmCorrelationExpression) {
setConfirmCorrelationExpression(this.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;
}
public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
this.defaultDeliveryMode = defaultDeliveryMode;
}
/**
* Set to {@code false} to attempt to connect during endpoint start;
* default {@code true}, meaning the connection will be attempted
* to be established on the arrival of the first message.
* @param lazyConnect the lazyConnect to set
* @since 4.1
*/
public void setLazyConnect(boolean lazyConnect) {
this.lazyConnect = lazyConnect;
setConfirmCorrelationExpression(confirmCorrelationExpression);
}
@Override
@@ -228,45 +89,13 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
@Override
protected void doInit() {
Assert.state(exchangeNameExpression == null || exchangeName == null,
"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
BeanFactory beanFactory = this.getBeanFactory();
if (this.exchangeNameExpression != null) {
this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(this.exchangeNameExpression,
String.class);
if (beanFactory != null) {
this.exchangeNameGenerator.setBeanFactory(beanFactory);
}
}
Assert.state(routingKeyExpression == null || routingKey == null,
"Either a routingKey or a routingKeyExpression can be provided, but not both");
if (this.routingKeyExpression != null) {
this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor<String>(this.routingKeyExpression,
String.class);
if (beanFactory != null) {
this.routingKeyGenerator.setBeanFactory(beanFactory);
}
}
if (this.confirmCorrelationExpression != null) {
this.correlationDataGenerator =
new ExpressionEvaluatingMessageProcessor<Object>(this.confirmCorrelationExpression, Object.class);
protected void endpointInit() {
if (getConfirmCorrelationExpression() != null) {
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
"RabbitTemplate implementation is required for publisher confirms");
((RabbitTemplate) this.amqpTemplate).setConfirmCallback(this);
if (beanFactory != null) {
this.correlationDataGenerator.setBeanFactory(beanFactory);
}
}
else {
NullChannel nullChannel = extractTypeIfPossible(this.confirmAckChannel, NullChannel.class);
Assert.state(this.confirmAckChannel == null || nullChannel != null,
"A 'confirmCorrelationExpression' is required when specifying a 'confirmAckChannel'");
nullChannel = extractTypeIfPossible(this.confirmNackChannel, NullChannel.class);
Assert.state(this.confirmNackChannel == null || nullChannel != null,
"A 'confirmCorrelationExpression' is required when specifying a 'confirmNackChannel'");
}
if (this.returnChannel != null) {
if (getReturnChannel() != null) {
Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
"RabbitTemplate implementation is required for publisher confirms");
((RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
@@ -274,64 +103,17 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
if (!this.lazyConnect && event.getApplicationContext().equals(getApplicationContext())
&& this.amqpTemplate instanceof RabbitTemplate) {
ConnectionFactory connectionFactory = ((RabbitTemplate) this.amqpTemplate).getConnectionFactory();
if (connectionFactory != null) {
try {
Connection connection = connectionFactory.createConnection();
if (connection != null) {
connection.close();
}
}
catch (RuntimeException e) {
logger.error("Failed to eagerly establish the connection.", e);
}
}
}
}
@Override
public void start() {
}
@Override
public void stop() {
protected void doStop() {
if (this.amqpTemplate instanceof Lifecycle) {
((Lifecycle) this.amqpTemplate).stop();
}
}
@Override
public boolean isRunning() {
return !(this.amqpTemplate instanceof Lifecycle) || ((Lifecycle) this.amqpTemplate).isRunning();
}
@Override
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);
}
if (this.routingKeyGenerator != null) {
routingKey = this.routingKeyGenerator.processMessage(requestMessage);
}
CorrelationData correlationData = generateCorrelationData(requestMessage);
String exchangeName = generateExchangeName(requestMessage);
String routingKey = generateRoutingKey(requestMessage);
if (this.expectReply) {
return this.sendAndReceive(exchangeName, routingKey, requestMessage, correlationData);
}
@@ -354,7 +136,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
@Override
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(),
getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(),
message.getMessageProperties());
return message;
}
@@ -375,76 +157,12 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
if (amqpReplyMessage == null) {
return null;
}
Object replyObject = converter.fromMessage(amqpReplyMessage);
AbstractIntegrationMessageBuilder<?> builder = (replyObject instanceof Message)
? this.getMessageBuilderFactory().fromMessage((Message<?>) replyObject)
: this.getMessageBuilderFactory().withPayload(replyObject);
Map<String, ?> headers = this.headerMapper.toHeadersFromReply(amqpReplyMessage.getMessageProperties());
builder.copyHeadersIfAbsent(headers);
return builder.build();
}
protected org.springframework.amqp.core.Message mapMessage(Message<?> requestMessage, MessageConverter converter) {
MessageProperties amqpMessageProperties = new MessageProperties();
org.springframework.amqp.core.Message amqpMessage;
if (converter instanceof ContentTypeDelegatingMessageConverter) {
this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
}
else { // See INT-3002 - map headers last if we're not using a CTDMC
amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
}
checkDeliveryMode(requestMessage, amqpMessageProperties);
return amqpMessage;
}
private void checkDeliveryMode(Message<?> requestMessage, MessageProperties messageProperties) {
if (this.defaultDeliveryMode != null &&
requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) {
messageProperties.setDeliveryMode(this.defaultDeliveryMode);
}
return buildReplyMessage(converter, amqpReplyMessage);
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
Object userCorrelationData = correlationData;
if (correlationData == null) {
if (logger.isDebugEnabled()) {
logger.debug("No correlation data provided for ack: " + ack + " cause:" + cause);
}
return;
}
if (correlationData instanceof CorrelationDataWrapper) {
userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData();
}
Map<String, Object> headers = new HashMap<String, Object>();
headers.put(AmqpHeaders.PUBLISH_CONFIRM, ack);
if (!ack && StringUtils.hasText(cause)) {
headers.put(AmqpHeaders.PUBLISH_CONFIRM_NACK_CAUSE, cause);
}
AbstractIntegrationMessageBuilder<?> builder = userCorrelationData instanceof Message
? this.getMessageBuilderFactory().fromMessage((Message<?>) userCorrelationData)
: this.getMessageBuilderFactory().withPayload(userCorrelationData);
Message<?> confirmMessage = builder
.copyHeaders(headers)
.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);
}
}
handleConfirm(correlationData, ack, cause);
}
@Override
@@ -452,33 +170,9 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
String exchange, String routingKey) {
// safe to cast; we asserted we have a RabbitTemplate in doInit()
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
Object returnedObject = converter.fromMessage(message);
AbstractIntegrationMessageBuilder<?> builder = (returnedObject instanceof Message)
? this.getMessageBuilderFactory().fromMessage((Message<?>) returnedObject)
: this.getMessageBuilderFactory().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());
}
private static class CorrelationDataWrapper extends CorrelationData {
private final Object userData;
private CorrelationDataWrapper(String id, Object userData) {
super(id);
this.userData = userData;
}
public Object getUserData() {
return this.userData;
}
Message<?> returned = buildReturnedMessage(message, replyCode, replyText, exchange,
routingKey, converter);
getReturnChannel().send(returned);
}
}

186
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java Normal file
View File

@@ -0,0 +1,186 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.amqp.outbound;
import org.springframework.amqp.core.AmqpMessageReturnedException;
import org.springframework.amqp.core.AmqpReplyTimeoutException;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.integration.handler.ReplyRequiredException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* An outbound gateway where the sending thread is released immediately and the reply
* is sent on the async template's listener container thread.
*
* @author Gary Russell
* @since 4.3
*
*/
public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
private final AsyncRabbitTemplate template;
private final MessageConverter messageConverter;
public AsyncAmqpOutboundGateway(AsyncRabbitTemplate template) {
Assert.notNull(template, "AsyncRabbitTemplate cannot be null");
this.template = template;
this.messageConverter = template.getMessageConverter();
Assert.notNull(this.messageConverter, "the template's message converter cannot be null");
setConnectionFactory(this.template.getConnectionFactory());
setAsyncReplySupported(true);
}
@Override
public String getComponentType() {
return "amqp:outbound-async-gateway";
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
RabbitMessageFuture future = this.template.sendAndReceive(generateExchangeName(requestMessage),
generateRoutingKey(requestMessage), mapMessage(requestMessage, this.messageConverter));
future.addCallback(new FutureCallback(requestMessage));
CorrelationData correlationData = generateCorrelationData(requestMessage);
if (correlationData != null && future.getConfirm() != null) {
future.getConfirm().addCallback(new CorrelationCallback(correlationData, future));
}
return null;
}
private final class FutureCallback implements ListenableFutureCallback<org.springframework.amqp.core.Message> {
private final Message<?> requestMessage;
FutureCallback(Message<?> requestMessage) {
this.requestMessage = requestMessage;
}
@Override
public void onSuccess(org.springframework.amqp.core.Message result) {
Message<?> replyMessage = null;
try {
replyMessage = buildReplyMessage(AsyncAmqpOutboundGateway.this.messageConverter, result);
sendOutputs(replyMessage, this.requestMessage);
}
catch (Exception e) {
Exception exceptionToLogAndSend = e;
if (!(e instanceof MessagingException)) {
exceptionToLogAndSend = new MessageHandlingException(this.requestMessage, e);
if (replyMessage != null) {
exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
}
}
logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend);
sendErrorMessage(exceptionToLogAndSend, this.requestMessage.getHeaders().getErrorChannel());
}
}
@Override
public void onFailure(Throwable ex) {
Throwable exceptionToSend = ex;
if (ex instanceof AmqpReplyTimeoutException) {
if (getRequiresReply()) {
exceptionToSend = new ReplyRequiredException(this.requestMessage, "Timeout on async request/reply",
ex);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Reply not required and async timeout for " + this.requestMessage);
}
return;
}
}
if (ex instanceof AmqpMessageReturnedException) {
if (getReturnChannel() == null) {
logger.error("Returned message received and no return channel "
+ ((AmqpMessageReturnedException) ex).getReturnedMessage());
}
else {
AmqpMessageReturnedException amre = (AmqpMessageReturnedException) ex;
Message<?> returnedMessage = buildReturnedMessage(
amre.getReturnedMessage(), amre.getReplyCode(), amre.getReplyText(), amre.getExchange(),
amre.getRoutingKey(), AsyncAmqpOutboundGateway.this.messageConverter);
sendOutput(returnedMessage, getReturnChannel(), true);
}
}
else {
sendErrorMessage(exceptionToSend, this.requestMessage.getHeaders().getErrorChannel());
}
}
private void sendErrorMessage(Throwable ex, Object errorChannel) {
Throwable result = ex;
if (!(ex instanceof MessagingException)) {
result = new MessageHandlingException(this.requestMessage, ex);
}
if (errorChannel == null) {
logger.error("Async exception received and no 'errorChannel' header exists; cannot route "
+ "exception to caller", result);
}
else {
try {
sendOutput(new ErrorMessage(result), errorChannel, true);
}
catch (Exception e) {
Exception exceptionToLog = e;
if (!(e instanceof MessagingException)) {
exceptionToLog = new MessageHandlingException(this.requestMessage, e);
}
logger.error("Failed to send async reply", exceptionToLog);
}
}
}
}
private final class CorrelationCallback implements ListenableFutureCallback<Boolean> {
private final CorrelationData correlationData;
private final RabbitMessageFuture replyFuture;
CorrelationCallback(CorrelationData correlationData, RabbitMessageFuture replyFuture) {
this.correlationData = correlationData;
this.replyFuture = replyFuture;
}
@Override
public void onSuccess(Boolean result) {
try {
handleConfirm(this.correlationData, result, this.replyFuture.getNackCause());
}
catch (Exception e) {
logger.error("Failed to send publisher confirm");
}
}
@Override
public void onFailure(Throwable ex) {
}
}
}

204
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd
View File

@@ -43,6 +43,19 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="amqp-template" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
An AmqpTemplate; if not supplied, a bean name 'amqpTemplate' is
expected.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -80,91 +93,55 @@
<xsd:annotation>
<xsd:documentation>
Configures a gateway that will publish an AMQP Message to the provided Exchange
and expect a reply Message.
</xsd:documentation>
and expect a reply Message. The thread blocks waiting for a reply or timeout; uses
'RabbitTemplate.sendAndReceive()'.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="outboundType">
<xsd:attribute name="id" type="xsd:string">
<xsd:extension base="outboundGatewayType">
<xsd:attribute name="amqp-template" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Unique ID for this gateway.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="request-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.
</xsd:documentation>
<xsd:appinfo>
<xsd:documentation>
An AmqpTemplate; if not supplied, a bean name 'amqpTemplate' is
expected.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.messaging.MessageChannel"/>
<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-channel" type="xsd:string">
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-async-gateway">
<xsd:annotation>
<xsd:documentation>
Configures a gateway that will publish an AMQP Message to the provided Exchange
and expect a reply Message. The sending thread returns immediately; the reply
is sent asynchronously; uses 'AsyncRabbitTemplate.sendAndReceive()'.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="outboundGatewayType">
<xsd:attribute name="async-template" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Message Channel to which replies should be sent after being received from an AMQP Queue and converted.
An AsyncRabbitTemplate; if not supplied, a bean name 'asyncRabbitTemplate' is
expected.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.messaging.MessageChannel"/>
<tool:expected-type type="org.springframework.amqp.rabbit.core.AsyncRabbitTemplate" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to specify how long this gateway will wait for
the reply message to be sent successfully to the reply channel
before throwing an exception. This attribute only applies when the
channel might block, for example when using a bounded queue channel that
is currently full.
Also, keep in mind that when sending to a DirectChannel, the
invocation will occur in the sender's thread. Therefore,
the failing of the send operation may be caused by other
components further downstream.
The "reply-timeout" attribute maps to the "sendTimeout" property of the
underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
The attribute will default, if not specified, to '-1', meaning that
by default, the Gateway will wait indefinitely. The value is
specified in milliseconds.
]]></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").
A special token 'STANDARD_REPLY_HEADERS' represents all the standard AMQP headers (replyTo, correlationId etc);
it is included by default. If you wish to add your own headers, you must also include this token if you wish the
standard headers to also be mapped. To map all non-standard headers the 'NON_STANDARD_HEADERS' token can be used.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="requires-reply" type="xsd:string" use="optional" default="true">
<xsd:annotation>
<xsd:documentation>
Specify whether this outbound gateway must return a non-null value. This value is
'true' by default, and a ReplyRequiredException will be thrown when
the underlying service returns a null value.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -401,15 +378,6 @@ The routing-key to use when sending Messages evaluated as an expression on the m
</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>
@@ -530,6 +498,92 @@ property set to TRUE.
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:complexType>
<xsd:complexType name="outboundGatewayType">
<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" 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.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="requires-reply" type="xsd:string" use="optional" default="true">
<xsd:annotation>
<xsd:documentation>
Specify whether this outbound gateway must return a non-null value. This value is
'true' by default, and a ReplyRequiredException will be thrown when
the underlying service returns a null value.
</xsd:documentation>
</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 AMQP Queue and converted.
</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="reply-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Allows you to specify how long this gateway will wait for
the reply message to be sent successfully to the reply channel
before throwing an exception. This attribute only applies when the
channel might block, for example when using a bounded queue channel that
is currently full.
Also, keep in mind that when sending to a DirectChannel, the
invocation will occur in the sender's thread. Therefore,
the failing of the send operation may be caused by other
components further downstream.
The "reply-timeout" attribute maps to the "sendTimeout" property of the
underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
The attribute will default, if not specified, to '-1', meaning that
by default, the Gateway will wait indefinitely. The value is
specified in milliseconds.
]]></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").
A special token 'STANDARD_REPLY_HEADERS' represents all the standard AMQP headers (replyTo, correlationId etc);
it is included by default. If you wish to add your own headers, you must also include this token if you wish the
standard headers to also be mapped. To map all non-standard headers the 'NON_STANDARD_HEADERS' token can be used.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
<xsd:simpleType name="deliveryModeEnumeration">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="PERSISTENT" />

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -60,7 +60,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
@@ -320,12 +319,13 @@ public class AmqpOutboundChannelAdapterParserTests {
handler.setApplicationContext(context);
handler.setBeanFactory(context);
handler.afterPropertiesSet();
ContextRefreshedEvent event = new ContextRefreshedEvent(context);
handler.onApplicationEvent(event);
handler.start();
handler.stop();
verify(logger, never()).error(Matchers.anyString(), any(RuntimeException.class));
handler.setLazyConnect(false);
handler.onApplicationEvent(event);
handler.start();
verify(logger).error("Failed to eagerly establish the connection.", toBeThrown);
handler.stop();
}

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

@@ -85,4 +85,31 @@
mapped-reply-headers=""/>
</int:chain>
<amqp:outbound-async-gateway id="asyncGateway" request-channel="toRabbit0"
reply-channel="fromRabbit"
reply-timeout="777"
requires-reply="false"
exchange-name="si.test.exchange"
routing-key="si.test.binding"
async-template="asyncTemplate"
auto-startup="false"
order="5"
return-channel="returnChannel">
<int:poller fixed-delay="100"/>
</amqp:outbound-async-gateway>
<bean id="asyncTemplate" class="org.springframework.amqp.rabbit.AsyncRabbitTemplate">
<constructor-arg>
<bean class="org.springframework.amqp.rabbit.core.RabbitTemplate">
<constructor-arg ref="connectionFactory" />
</bean>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
<constructor-arg ref="connectionFactory" />
<property name="queueNames" value="replies" />
</bean>
</constructor-arg>
</bean>
</beans>

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

@@ -42,7 +42,9 @@ import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.outbound.AsyncAmqpOutboundGateway;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.Orderable;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
@@ -70,10 +72,21 @@ public class AmqpOutboundGatewayParserTests {
Object edc = context.getBean("rabbitGateway");
assertFalse(TestUtils.getPropertyValue(edc, "autoStartup", Boolean.class));
AmqpOutboundEndpoint gateway = TestUtils.getPropertyValue(edc, "handler", AmqpOutboundEndpoint.class);
assertEquals(5, gateway.getOrder());
assertTrue(TestUtils.getPropertyValue(gateway, "requiresReply", Boolean.class));
assertEquals(context.getBean("fromRabbit"), TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals("amqp:outbound-gateway", gateway.getComponentType());
assertTrue(TestUtils.getPropertyValue(gateway, "requiresReply", Boolean.class));
checkGWProps(context, gateway);
AsyncAmqpOutboundGateway async = context.getBean("asyncGateway.handler", AsyncAmqpOutboundGateway.class);
assertEquals("amqp:outbound-async-gateway", async.getComponentType());
checkGWProps(context, async);
assertSame(context.getBean("asyncTemplate"), TestUtils.getPropertyValue(async, "template"));
context.close();
}
protected void checkGWProps(ConfigurableApplicationContext context, Orderable gateway) {
assertEquals(5, gateway.getOrder());
assertEquals(context.getBean("fromRabbit"), TestUtils.getPropertyValue(gateway, "outputChannel"));
MessageChannel returnChannel = context.getBean("returnChannel", MessageChannel.class);
assertSame(returnChannel, TestUtils.getPropertyValue(gateway, "returnChannel"));
@@ -81,8 +94,6 @@ public class AmqpOutboundGatewayParserTests {
assertEquals(Long.valueOf(777), sendTimeout);
assertTrue(TestUtils.getPropertyValue(gateway, "lazyConnect", Boolean.class));
context.close();
}
@SuppressWarnings("rawtypes")

221
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java Normal file
View File

@@ -0,0 +1,221 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.amqp.outbound;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.AfterClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.amqp.core.AmqpReplyTimeoutException;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.amqp.rule.BrokerRunning;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.concurrent.SettableListenableFuture;
/**
* @author Gary Russell
* @since 4.3
*
*/
public class AsyncAmqpGatewayTests {
@ClassRule
public static BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues("asyncQ1", "asyncRQ1");
// @Rule
// public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, "org.springframework.integration",
// "org.springframework.amqp");
@AfterClass
public static void tearDown() {
brokerRunning.removeTestQueues();
}
@Test
public void testConfirmsAndReturns() {
CachingConnectionFactory ccf = new CachingConnectionFactory("localhost");
ccf.setPublisherConfirms(true);
ccf.setPublisherReturns(true);
RabbitTemplate template = new RabbitTemplate(ccf);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(ccf);
container.setBeanName("replyContainer");
container.setQueueNames("asyncRQ1");
container.afterPropertiesSet();
container.start();
AsyncRabbitTemplate asyncTemplate = spy(new AsyncRabbitTemplate(template, container));
asyncTemplate.setEnableConfirms(true);
asyncTemplate.setMandatory(true);
asyncTemplate.start();
SimpleMessageListenerContainer receiver = new SimpleMessageListenerContainer(ccf);
receiver.setBeanName("receiver");
receiver.setQueueNames("asyncQ1");
final CountDownLatch waitForAckBeforeReplying = new CountDownLatch(1);
MessageListenerAdapter messageListener = new MessageListenerAdapter(new Object() {
@SuppressWarnings("unused")
public String handleMessage(String foo) {
try {
waitForAckBeforeReplying.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return foo.toUpperCase();
}
});
receiver.setMessageListener(messageListener);
receiver.afterPropertiesSet();
receiver.start();
AsyncAmqpOutboundGateway gateway = new AsyncAmqpOutboundGateway(asyncTemplate);
QueueChannel outputChannel = new QueueChannel();
outputChannel.setBeanName("output");
QueueChannel returnChannel = new QueueChannel();
returnChannel.setBeanName("returns");
QueueChannel ackChannel = new QueueChannel();
ackChannel.setBeanName("acks");
QueueChannel nackChannel = new QueueChannel();
nackChannel.setBeanName("nacks");
QueueChannel errorChannel = new QueueChannel();
errorChannel.setBeanName("errors");
gateway.setOutputChannel(outputChannel);
gateway.setReturnChannel(returnChannel);
gateway.setConfirmAckChannel(ackChannel);
gateway.setConfirmNackChannel(nackChannel);
gateway.setConfirmCorrelationExpressionString("#this");
gateway.setExchangeName("");
gateway.setRoutingKey("asyncQ1");
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo").setErrorChannel(errorChannel).build();
gateway.handleMessage(message);
Message<?> ack = ackChannel.receive(10000);
assertNotNull(ack);
assertEquals("foo", ack.getPayload());
assertEquals(true, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
waitForAckBeforeReplying.countDown();
Message<?> received = outputChannel.receive(10000);
assertNotNull(received);
assertEquals("FOO", received.getPayload());
// timeout
asyncTemplate.setReceiveTimeout(100);
receiver.setMessageListener(new MessageListener() {
@Override
public void onMessage(org.springframework.amqp.core.Message message) {
}
});
gateway.handleMessage(message);
assertNull(errorChannel.receive(1000));
gateway.setRequiresReply(true);
gateway.handleMessage(message);
received = errorChannel.receive(10000);
assertThat(received, instanceOf(ErrorMessage.class));
ErrorMessage error = (ErrorMessage) received;
assertThat(error.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) error.getPayload()).getCause(), instanceOf(AmqpReplyTimeoutException.class));
asyncTemplate.setReceiveTimeout(30000);
receiver.setMessageListener(messageListener);
// error on sending result
DirectChannel errorForce = new DirectChannel();
errorForce.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("intentional");
}
});
gateway.setOutputChannel(errorForce);
gateway.handleMessage(message);
received = errorChannel.receive(10000);
assertThat(received, instanceOf(ErrorMessage.class));
error = (ErrorMessage) received;
assertThat(error.getPayload(), instanceOf(MessagingException.class));
assertEquals("FOO", ((MessagingException) error.getPayload()).getFailedMessage().getPayload());
gateway.setRoutingKey(UUID.randomUUID().toString());
gateway.handleMessage(message);
Message<?> returned = returnChannel.receive(10000);
assertNotNull(returned);
assertEquals("foo", returned.getPayload());
// Simulate a nack - hard to get Rabbit to generate one
RabbitMessageFuture future = asyncTemplate.new RabbitMessageFuture(null, null);
doReturn(future).when(asyncTemplate).sendAndReceive(anyString(), anyString(),
any(org.springframework.amqp.core.Message.class));
DirectFieldAccessor dfa = new DirectFieldAccessor(future);
dfa.setPropertyValue("nackCause", "nacknack");
SettableListenableFuture<Boolean> confirmFuture = new SettableListenableFuture<Boolean>();
confirmFuture.set(false);
dfa.setPropertyValue("confirm", confirmFuture);
gateway.handleMessage(message);
ack = nackChannel.receive(10000);
assertNotNull(ack);
assertEquals("foo", ack.getPayload());
assertEquals("nacknack", ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM_NACK_CAUSE));
assertEquals(false, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
asyncTemplate.stop();
receiver.stop();
ccf.destroy();
}
}

9
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java
View File

@@ -132,4 +132,13 @@ public class BrokerRunning extends TestWatcher {
return super.apply(base, description);
}
public void removeTestQueues() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
RabbitAdmin admin = new RabbitAdmin(connectionFactory);
for (Queue queue : this.queues) {
admin.deleteQueue(queue.getName());
}
connectionFactory.destroy();
}
}

29
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -32,6 +32,7 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
@@ -82,10 +83,19 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
*
* @since 4.3
*/
protected void setAsyncReplySupported(boolean asyncReplySupported) {
protected final void setAsyncReplySupported(boolean asyncReplySupported) {
this.asyncReplySupported = asyncReplySupported;
}
/**
* @see #setAsyncReplySupported(boolean)
* @return true if this handler supports async replies.
* @since 4.3
*/
protected boolean getAsyncReplySupported() {
return this.asyncReplySupported;
}
@Override
protected void onInit() throws Exception {
super.onInit();
@@ -174,13 +184,18 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
@Override
public void onSuccess(Object result) {
Message<?> replyMessage = null;
try {
sendOutput(createOutputMessage(result, requestHeaders), theReplyChannel, false);
replyMessage = createOutputMessage(result, requestHeaders);
sendOutput(replyMessage, theReplyChannel, false);
}
catch (Exception e) {
Exception exceptionToLogAndSend = e;
if (!(e instanceof MessagingException)) {
exceptionToLogAndSend = new MessageHandlingException(requestMessage, e);
if (replyMessage != null) {
exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
}
}
logger.error("Failed to send async reply: " + result.toString(), exceptionToLogAndSend);
onFailure(exceptionToLogAndSend);
@@ -200,7 +215,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
}
else {
try {
sendOutput(createOutputMessage(result, requestHeaders), errorChannel, true);
sendOutput(new ErrorMessage(result), errorChannel, true);
}
catch (Exception e) {
Exception exceptionToLog = e;
@@ -254,7 +269,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
}
}
private Message<?> createOutputMessage(Object output, MessageHeaders requestHeaders) {
protected Message<?> createOutputMessage(Object output, MessageHeaders requestHeaders) {
AbstractIntegrationMessageBuilder<?> builder = null;
if (output instanceof Message<?>) {
if (!this.shouldCopyRequestHeaders()) {
@@ -280,12 +295,12 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
* <code>null</code>, and it must be an instance of either String or {@link MessageChannel}.
* @param output the output object to send
* @param replyChannel the 'replyChannel' value from the original request
* @param isError - this is an error, use the replyChannel argument (must not be null), not
* @param useArgChannel - use the replyChannel argument (must not be null), not
* the configured output channel.
*/
private void sendOutput(Object output, Object replyChannel, boolean isError) {
protected void sendOutput(Object output, Object replyChannel, boolean useArgChannel) {
MessageChannel outputChannel = getOutputChannel();
if (!isError && outputChannel != null) {
if (!useArgChannel && outputChannel != null) {
replyChannel = outputChannel;
}
if (replyChannel == null) {

7
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

@@ -57,6 +57,9 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
this.requiresReply = requiresReply;
}
protected boolean getRequiresReply() {
return requiresReply;
}
public void setAdviceChain(List<Advice> adviceChain) {
Assert.notNull(adviceChain, "adviceChain cannot be null");
@@ -104,11 +107,11 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
if (result != null) {
sendOutputs(result, message);
}
else if (this.requiresReply) {
else if (this.requiresReply && !getAsyncReplySupported()) {
throw new ReplyRequiredException(message, "No reply produced by handler '" +
getComponentName() + "', and its 'requiresReply' property is set to true.");
}
else if (logger.isDebugEnabled()) {
else if (!getAsyncReplySupported() && logger.isDebugEnabled()) {
logger.debug("handler '" + this + "' produced no reply for request Message: " + message);
}
}

19
spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyRequiredException.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2016 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.
@@ -22,15 +22,30 @@ import org.springframework.messaging.MessagingException;
/**
* Exception that indicates no reply message is produced by a handler
* that does have a value of true for the 'requiresReply' property.
*
*
* @author Mark Fisher
* @author Gary Russell
* @since 2.1
*/
@SuppressWarnings("serial")
public class ReplyRequiredException extends MessagingException {
/**
* @param failedMessage the failed message.
* @param description the description.
*/
public ReplyRequiredException(Message<?> failedMessage, String description) {
super(failedMessage, description);
}
/**
* @param failedMessage the failed message.
* @param description the description.
* @param t the root cause.
* @since 4.3
*/
public ReplyRequiredException(Message<?> failedMessage, String description, Throwable t) {
super(failedMessage, description, t);
}
}

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

@@ -11,6 +11,7 @@ The following adapters are available:
* <<amqp-inbound-gateway,Inbound Gateway>>
* <<amqp-outbound-channel-adapter,Outbound Channel Adapter>>
* <<amqp-outbound-gateway,Outbound Gateway>>
* <<amqp-async-outbound-gateway,Async Outbound Gateway>>
Spring Integration also provides a point-to-point Message Channel as well as a publish/subscribe Message Channel backed by AMQP Exchanges and Queues.
@@ -529,7 +530,7 @@ Mutually exclusive with 'exchange-name'.
_Optional_.
<6> The order for this consumer when multiple consumers are registered thereby enabling load- balancing and/or failover.
<6> The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover.
_Optional (Defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE])_.
@@ -545,11 +546,12 @@ Mutually exclusive with 'routing-key'.
_Optional_.
<9> The default delivery mode for messages; 'PERSISTENT' or 'NON_PERSISTENT'.
Overridden if the 'header-mapper' sets the delivery mode.
The 'DefaultHeaderMapper' sets the value if the Spring Integration message header `amqp_deliveryMode` is present.
If this attribute is not supplied and the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter' used by the 'RabbitTemplate'.
If that is not customized at all, the default is 'PERSISTENT'.
<9> The default delivery mode for messages; `PERSISTENT` or `NON_PERSISTENT`.
Overridden if the `header-mapper` sets the delivery mode.
The `DefaultHeaderMapper` sets the value if the Spring Integration message header `amqp_deliveryMode` is present.
If this attribute is not supplied and the header mapper doesn't set it, the default depends on the underlying
spring-amqp `MessagePropertiesConverter` used by the `RabbitTemplate`.
If that is not customized at all, the default is `PERSISTENT`.
_Optional_.
@@ -693,7 +695,7 @@ public class AmqpJavaApplication {
[[amqp-outbound-gateway]]
=== Outbound Gateway
A configuration sample for an AMQP Outbound Gateway is shown below.
Configuration for an AMQP Outbound Gateway is shown below.
[source,xml]
----
@@ -740,11 +742,11 @@ Mutually exclusive with 'exchange-name'.
_Optional_.
<6> The order for this consumer when multiple consumers are registered thereby enabling load- balancing and/or failover.
<6> The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover.
_Optional (Defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE])_.
<7> Message Channel to which replies should be sent after being received from an AQMP Queue and converted._Optional_.
<7> Message Channel to which replies should be sent after being received from an AMQP Queue and converted._Optional_.
<8> The time the gateway will wait when sending the reply message to the `reply-channel`.
@@ -769,11 +771,12 @@ Mutually exclusive with 'routing-key'.
_Optional_.
<12> The default delivery mode for messages; 'PERSISTENT' or 'NON_PERSISTENT'.
Overridden if the 'header-mapper' sets the delivery mode.
The 'DefaultHeaderMapper' sets the value if the Spring Integration message header `amqp_deliveryMode` is present.
If this attribute is not supplied and the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter' used by the 'RabbitTemplate'.
If that is not customized at all, the default is 'PERSISTENT'.
<12> The default delivery mode for messages; `PERSISTENT` or `NON_PERSISTENT`.
Overridden if the `header-mapper` sets the delivery mode.
The `DefaultHeaderMapper` sets the value if the Spring Integration message header `amqp_deliveryMode` is present.
If this attribute is not supplied and the header mapper doesn't set it, the default depends on the underlying
spring-amqp `MessagePropertiesConverter` used by the `RabbitTemplate`.
If that is not customized at all, the default is `PERSISTENT`.
_Optional_.
<13> Since _version 4.2_. An expression defining correlation data.
@@ -803,7 +806,7 @@ _Optional_.
<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.
This allows "fail fast" detection of bad configuration, by logging an error message 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.
@@ -906,6 +909,188 @@ public class AmqpJavaApplication {
}
----
[[amqp-async-outbound-gateway]]
=== Async Outbound Gateway
The gateway discussed in the previous section is synchronous, in that the sending thread is suspended until a
reply is received (or a timeout occurs).
Spring Integration _version 4.3_ added this asynchronous gateway, which uses the `AsyncRabbitTemplate` from Spring AMQP.
When a message is sent, the thread returns immediately and the reply is sent on the template's listener container
thread when it is received.
Configuration for an AMQP Async Outbound Gateway is shown below.
[source,xml]
----
<int-amqp:outbound-gateway id="inboundGateway" <1>
request-channel="myRequestChannel" <2>
async-template="" <3>
exchange-name="" <4>
exchange-name-expression="" <5>
order="1" <6>
reply-channel="" <7>
reply-timeout="" <8>
requires-reply="" <9>
routing-key="" <10>
routing-key-expression="" <11>
default-delivery-mode"" <12>
confirm-correlation-expression="" <13>
confirm-ack-channel="" <14>
confirm-nack-channel="" <15>
return-channel="" <16>
lazy-connect="true" /> <17>
----
<1> Unique ID for this adapter.
_Optional_.
<2> Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
_Required_.
<3> Bean Reference to the configured `AsyncRabbitTemplate` _Optional (Defaults to "asyncRabbitTemplate")_.
<4> 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.
Mutually exclusive with 'exchange-name-expression'.
_Optional_.
<5> A SpEL expression that is evaluated to determine the name of the AMQP Exchange to which Messages should be sent,
with the message as the root object.
If not provided, Messages will be sent to the default, no-name Exchange.
Mutually exclusive with 'exchange-name'.
_Optional_.
<6> The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover.
_Optional (Defaults to Ordered.LOWEST_PRECEDENCE [=Integer.MAX_VALUE])_.
<7> Message Channel to which replies should be sent after being received from an AMQP Queue and converted. _Optional_.
<8> The time the gateway will wait when sending the reply message to the `reply-channel`.
This only applies if the `reply-channel` can block - such as a `QueueChannel` with a capacity limit that is currently full.
Default: infinity.
<9> When `true`, the gateway will send an error message to the inbound message's `errorChannel` header if no reply
message is received within the `AsyncRabbitTemplate`'s `receiveTimeout` property. Default: `true`.
<10> The routing-key to use when sending Messages.
By default, this will be an empty String.
Mutually exclusive with 'routing-key-expression'.
_Optional_.
<11> A SpEL expression that is evaluated to determine the routing-key to use when sending Messages, with the message as the root object (e.g.
'payload.key').
By default, this will be an empty String.
Mutually exclusive with 'routing-key'.
_Optional_.
<12> The default delivery mode for messages; `PERSISTENT` or `NON_PERSISTENT`.
Overridden if the `header-mapper` sets the delivery mode.
The `DefaultHeaderMapper` sets the value if the Spring Integration message header `amqp_deliveryMode` is present.
If this attribute is not supplied and the header mapper doesn't set it, the default depends on the underlying
spring-amqp `MessagePropertiesConverter` used by the `RabbitTemplate`.
If that is not customized at all, the default is `PERSISTENT`.
_Optional_.
<13> 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).
For nacks, an additional header `amqp_publishConfirmNackCause` is provided.
Examples: "headers['myCorrelationData']", "payload".
If the expression resolves to a `Message<?>` instance (such as "`#this`"), the message
emitted on the ack/nack channel is based on that message, with the additional header(s) added.
_Optional_.
<14> The channel to which positive (ack) publisher confirms are sent; payload is the correlation
data defined by the _confirm-correlation-expression_.
Requires the underlying `AsyncRabbitTemplate` to have its `enableConfirms` property set to true.
_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_.
Requires the underlying `AsyncRabbitTemplate` to have its `enableConfirms` property set to true.
_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_.
Requires the underlying `AsyncRabbitTemplate` to have its `mandatory` property set to true.
_Optional_.
<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, by logging an error message 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.
[IMPORTANT]
.RabbitTemplate
=====
When using confirms and returns, it is recommended that the `RabbitTemplate` wired into the `AsyncRabbitTemplate` be
dedicated.
Otherwise, unexpected side-effects may be encountered.
=====
==== Configuring with Java Configuration
The following configuration provides an example of configuring the outbound gateway using Java configuration:
[source, java]
----
@Configuration
public class AmqpAsyncConfig {
@Bean
@ServiceActivator(inputChannel = "amqpOutboundChannel")
public AsyncAmqpOutboundGateway amqpOutbound(AmqpTemplate asyncTemplate) {
AsyncAmqpOutboundGateway outbound = new AsyncAmqpOutboundGateway(asyncTemplate);
outbound.setRoutingKey("foo"); // default exchange - route to queue 'foo'
return outbound;
}
@Bean
public AsyncRabbitTemplate asyncTemplate(RabbitTemplate rabbitTemplate,
SimpleMessageListenerContainer replyContainer) {
return new AsyncRabbitTemplate(rabbitTemplate, replyContainer);
}
@Bean
public SimpleMessageListenerContainer replyContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(ccf);
container.setQueueNames("asyncRQ1");
return container;
}
@Bean
public MessageChannel amqpOutboundChannel() {
return new DirectChannel();
}
}
----
==== Configuring with the Java DSL
The following Spring Boot application provides an example of configuring the outbound adapter using the Java DSL:
[source, java]
----
// To be supplied when the DSL Amqp factory class adds support for the async gateway.
----
[[content-type-conversion-outbound]]
=== Outbound Message Conversion

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

@@ -9,6 +9,10 @@ development process.
[[x4.3-new-components]]
=== New Components
==== AMQP Async Outbound Gateway
See <<amqp-async-outbound-gateway>>.
==== MessageGroupFactory
The new `MessageGroupFactory` strategy has been introduced to allow a control over `MessageGroup` instances