From d4e6615f82ebf6c71b656f647b0fc57f79e4b302 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 1 Mar 2016 15:54:53 -0500 Subject: [PATCH] INT-3943: AMQP Async Outbound Gateway 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` --- .../amqp/config/AmqpNamespaceHandler.java | 7 +- .../config/AmqpOutboundGatewayParser.java | 25 +- .../AbstractAmqpOutboundEndpoint.java | 473 ++++++++++++++++++ .../amqp/outbound/AmqpOutboundEndpoint.java | 344 +------------ .../outbound/AsyncAmqpOutboundGateway.java | 186 +++++++ .../config/spring-integration-amqp-4.3.xsd | 204 +++++--- ...AmqpOutboundChannelAdapterParserTests.java | 10 +- ...AmqpOutboundGatewayParserTests-context.xml | 27 + .../AmqpOutboundGatewayParserTests.java | 21 +- .../amqp/outbound/AsyncAmqpGatewayTests.java | 221 ++++++++ .../integration/amqp/rule/BrokerRunning.java | 9 + .../AbstractMessageProducingHandler.java | 29 +- .../AbstractReplyProducingMessageHandler.java | 7 +- .../handler/ReplyRequiredException.java | 19 +- src/reference/asciidoc/amqp.adoc | 215 +++++++- src/reference/asciidoc/whats-new.adoc | 4 + 16 files changed, 1357 insertions(+), 444 deletions(-) create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java index a5f854f725..d351f280ff 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java @@ -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()); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java index 8443244850..bfd4453847 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java @@ -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); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java new file mode 100644 index 0000000000..a015cd72be --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -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 routingKeyGenerator; + + private volatile ExpressionEvaluatingMessageProcessor exchangeNameGenerator; + + private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); + + private volatile Expression confirmCorrelationExpression; + + private volatile ExpressionEvaluatingMessageProcessor 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 getRoutingKeyGenerator() { + return this.routingKeyGenerator; + } + + protected ExpressionEvaluatingMessageProcessor getExchangeNameGenerator() { + return this.exchangeNameGenerator; + } + + protected AmqpHeaderMapper getHeaderMapper() { + return this.headerMapper; + } + + protected Expression getConfirmCorrelationExpression() { + return this.confirmCorrelationExpression; + } + + protected ExpressionEvaluatingMessageProcessor 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(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(this.routingKeyExpression, + String.class); + if (beanFactory != null) { + this.routingKeyGenerator.setBeanFactory(beanFactory); + } + } + if (this.confirmCorrelationExpression != null) { + this.correlationDataGenerator = + new ExpressionEvaluatingMessageProcessor(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 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 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 headers = new HashMap(); + 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; + } + + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index abc7900c75..d7e75c1601 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -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, 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 routingKeyGenerator; - - private volatile ExpressionEvaluatingMessageProcessor exchangeNameGenerator; - - private volatile AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); - - private volatile Expression confirmCorrelationExpression; - - private volatile ExpressionEvaluatingMessageProcessor 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(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(this.routingKeyExpression, - String.class); - if (beanFactory != null) { - this.routingKeyGenerator.setBeanFactory(beanFactory); - } - } - if (this.confirmCorrelationExpression != null) { - this.correlationDataGenerator = - new ExpressionEvaluatingMessageProcessor(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 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 headers = new HashMap(); - 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 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); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java new file mode 100644 index 0000000000..69fe031a52 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -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 { + + 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 { + + 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) { + } + + } + +} diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd index 7a0c74f795..16839a3aa8 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd @@ -43,6 +43,19 @@ + + + + + An AmqpTemplate; if not supplied, a bean name 'amqpTemplate' is + expected. + + + + + + + @@ -80,91 +93,55 @@ Configures a gateway that will publish an AMQP Message to the provided Exchange - and expect a reply Message. - + and expect a reply Message. The thread blocks waiting for a reply or timeout; uses + 'RabbitTemplate.sendAndReceive()'. + - - + + - - Unique ID for this gateway. - - - - - - - Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange. - + + An AmqpTemplate; if not supplied, a bean name 'amqpTemplate' is + expected. + - + - + + + + + + + + + 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()'. + + + + + + - 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. - + - - - - - - - - - - - - - - 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. - - - @@ -401,15 +378,6 @@ The routing-key to use when sending Messages evaluated as an expression on the m - - - - - - - - - @@ -530,6 +498,92 @@ property set to TRUE. + + + + + + + Unique ID for this gateway. + + + + + + + Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange. + + + + + + + + + + + + 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. + + + + + + + Message Channel to which replies should be sent after being received from an AMQP Queue and converted. + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java index 73d6490899..90575a39ae 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java @@ -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(); } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml index 3173f7fd5c..d674a90ec8 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml @@ -85,4 +85,31 @@ mapped-reply-headers=""/> + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index a3d72277a6..3b1c907295 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -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") diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java new file mode 100644 index 0000000000..8ec916af21 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java @@ -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 confirmFuture = new SettableListenableFuture(); + 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(); + } + +} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java index 600dc59f7f..1bbca00c47 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java @@ -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(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index a5fe8157d6..ab9812c839 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -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 * null, 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 6c8850fbe2..beb40f6aab 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -57,6 +57,9 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa this.requiresReply = requiresReply; } + protected boolean getRequiresReply() { + return requiresReply; + } public void setAdviceChain(List 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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyRequiredException.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyRequiredException.java index 54f3f99b11..425fc8da23 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyRequiredException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyRequiredException.java @@ -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); + } + } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index fa554d2c71..bde5ca4160 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -11,6 +11,7 @@ The following adapters are available: * <> * <> * <> +* <> 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] +---- + + 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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index beda975ad9..683cb81894 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -9,6 +9,10 @@ development process. [[x4.3-new-components]] === New Components +==== AMQP Async Outbound Gateway + +See <>. + ==== MessageGroupFactory The new `MessageGroupFactory` strategy has been introduced to allow a control over `MessageGroup` instances