From 02565d04013d5a372c386f5c0d0f210325830d50 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 20 Aug 2015 11:44:32 -0400 Subject: [PATCH] INT-3807: `AmqpInboundGateway` Improvements JIRA: https://jira.spring.io/browse/INT-3807 Addressing PR comments INT-3807: Polishing --- build.gradle | 2 +- .../amqp/config/AmqpInboundGatewayParser.java | 9 +- .../amqp/inbound/AmqpInboundGateway.java | 132 +++++++++++++----- .../config/spring-integration-amqp-4.2.xsd | 25 ++++ .../AmqpInboundGatewayParserTests-context.xml | 12 +- .../config/AmqpInboundGatewayParserTests.java | 6 +- .../amqp/inbound/InboundEndpointTests.java | 24 ++-- src/reference/asciidoc/amqp.adoc | 15 +- src/reference/asciidoc/whats-new.adoc | 8 ++ 9 files changed, 176 insertions(+), 57 deletions(-) diff --git a/build.gradle b/build.gradle index 9ba3ab3144..d446c596bd 100644 --- a/build.gradle +++ b/build.gradle @@ -131,7 +131,7 @@ subprojects { subproject -> tomcatVersion = "8.0.18" smack3Version = '3.2.1' smackVersion = '4.0.6' - springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.5.0.RC1' + springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.5.0.BUILD-SNAPSHOT' // springCloudClusterVersion = '1.0.0.BUILD-SNAPSHOT' springDataMongoVersion = '1.7.2.RELEASE' springDataRedisVersion = '1.5.2.RELEASE' diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java index e6f7d712ce..eaa6c9ca46 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,12 +22,14 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.amqp.inbound.AmqpInboundGateway; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; /** * Parser for the AMQP 'inbound-gateway' element. * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class AmqpInboundGatewayParser extends AbstractAmqpInboundAdapterParser { @@ -39,8 +41,13 @@ public class AmqpInboundGatewayParser extends AbstractAmqpInboundAdapterParser { @Override protected void configureChannels(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + String amqpTemplateRef = element.getAttribute("amqp-template"); + if (StringUtils.hasText(amqpTemplateRef)) { + builder.addConstructorArgReference(amqpTemplateRef); + } IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "request-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-reply-to"); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index 97dcf11ffe..bbbe3e2a28 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -21,6 +21,7 @@ import java.util.Map; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; @@ -52,29 +53,58 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { private final AbstractMessageListenerContainer messageListenerContainer; + private final AmqpTemplate amqpTemplate; + + private final boolean amqpTemplateExplicitlySet; + private volatile MessageConverter amqpMessageConverter = new SimpleMessageConverter(); private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); - private final RabbitTemplate amqpTemplate; - + private Address defaultReplyTo; public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer) { + this(listenerContainer, new RabbitTemplate(listenerContainer.getConnectionFactory()), false); + } + + /** + * Construct {@link AmqpInboundGateway} based on the provided {@link AbstractMessageListenerContainer} + * to receive request messages and {@link AmqpTemplate} to send replies. + * @param listenerContainer the {@link AbstractMessageListenerContainer} to receive AMQP messages. + * @param amqpTemplate the {@link AmqpTemplate} to send reply messages. + * @since 4.2 + */ + public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { + this(listenerContainer, amqpTemplate, true); + } + + private AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate, + boolean amqpTemplateExplicitlySet) { Assert.notNull(listenerContainer, "listenerContainer must not be null"); + Assert.notNull(amqpTemplate, "'amqpTemplate' must not be null"); Assert.isNull(listenerContainer.getMessageListener(), "The listenerContainer provided to an AMQP inbound Gateway " + "must not have a MessageListener configured since " + "the adapter needs to configure its own listener implementation."); this.messageListenerContainer = listenerContainer; this.messageListenerContainer.setAutoStartup(false); - this.amqpTemplate = new RabbitTemplate(this.messageListenerContainer.getConnectionFactory()); + this.amqpTemplate = amqpTemplate; + this.amqpTemplateExplicitlySet = amqpTemplateExplicitlySet; } + /** + * Specify the {@link MessageConverter} to convert request and reply to/from {@link Message}. + * If the {@link #amqpTemplate} is explicitly set, this {@link MessageConverter} + * isn't populated there. You must configure that external {@link #amqpTemplate}. + * @param messageConverter the {@link MessageConverter} to use. + */ public void setMessageConverter(MessageConverter messageConverter) { Assert.notNull(messageConverter, "MessageConverter must not be null"); this.amqpMessageConverter = messageConverter; - this.amqpTemplate.setMessageConverter(messageConverter); + if (!amqpTemplateExplicitlySet) { + ((RabbitTemplate) this.amqpTemplate).setMessageConverter(messageConverter); + } } public void setHeaderMapper(AmqpHeaderMapper headerMapper) { @@ -82,6 +112,26 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { this.headerMapper = headerMapper; } + /** + * The {@code defaultReplyTo} address with the form + * 
+	 * (exchange)/(routingKey)
+	 *
+ * or + * 
+	 * (queueName)
+	 *
+ * if the request message doesn't have a {@code replyTo} property. + * The second form uses the default exchange ("") and the queue name as + * the routing key. + * @param defaultReplyTo the default {@code replyTo} address to use. + * @since 4.2 + * @see Address + */ + public void setDefaultReplyTo(String defaultReplyTo) { + this.defaultReplyTo = new Address(defaultReplyTo); + } + @Override public String getComponentType() { return "amqp:inbound-gateway"; @@ -102,48 +152,60 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build(); final org.springframework.messaging.Message reply = sendAndReceiveMessage(request); if (reply != null) { - // TODO: fallback to a reply address property of this gateway Address replyTo; String replyToProperty = message.getMessageProperties().getReplyTo(); - // TODO: Use the Address.AMQ_RABBITMQ_REPLY_TO constant when 1.4.3 is the minimum - if (replyToProperty.startsWith("amq.rabbitmq.reply-to")) { - replyTo = new Address("", replyToProperty); - } - else { + if (replyToProperty != null) { replyTo = new Address(replyToProperty); } - Assert.notNull(replyTo, "The replyTo header must not be null on a " + - "request Message being handled by the AMQP inbound gateway."); - amqpTemplate.convertAndSend(replyTo.getExchangeName(), replyTo.getRoutingKey(), reply.getPayload(), - new MessagePostProcessor() { + else { + replyTo = AmqpInboundGateway.this.defaultReplyTo; + } - @Override - public Message postProcessMessage(Message message) throws AmqpException { - MessageProperties messageProperties = message.getMessageProperties(); - String contentEncoding = messageProperties.getContentEncoding(); - long contentLength = messageProperties.getContentLength(); - String contentType = messageProperties.getContentType(); - headerMapper.fromHeadersToReply(reply.getHeaders(), messageProperties); - // clear the replyTo from the original message since we are using it now - messageProperties.setReplyTo(null); - // reset the content-* properties as determined by the MessageConverter - if (StringUtils.hasText(contentEncoding)) { - messageProperties.setContentEncoding(contentEncoding); - } - messageProperties.setContentLength(contentLength); - if (contentType != null) { - messageProperties.setContentType(contentType); - } - return message; - } + MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { - }); + @Override + public Message postProcessMessage(Message message) throws AmqpException { + MessageProperties messageProperties = message.getMessageProperties(); + String contentEncoding = messageProperties.getContentEncoding(); + long contentLength = messageProperties.getContentLength(); + String contentType = messageProperties.getContentType(); + headerMapper.fromHeadersToReply(reply.getHeaders(), messageProperties); + // clear the replyTo from the original message since we are using it now + messageProperties.setReplyTo(null); + // reset the content-* properties as determined by the MessageConverter + if (StringUtils.hasText(contentEncoding)) { + messageProperties.setContentEncoding(contentEncoding); + } + messageProperties.setContentLength(contentLength); + if (contentType != null) { + messageProperties.setContentType(contentType); + } + return message; + } + + }; + + if (replyTo != null) { + amqpTemplate.convertAndSend(replyTo.getExchangeName(), replyTo.getRoutingKey(), + reply.getPayload(), messagePostProcessor); + } + else { + if (!amqpTemplateExplicitlySet) { + throw new IllegalStateException("There is no 'replyTo' message property " + + "and the `defaultReplyTo` hasn't been configured."); + } + else { + amqpTemplate.convertAndSend(reply.getPayload(), messagePostProcessor); + } + } } } }); this.messageListenerContainer.afterPropertiesSet(); - this.amqpTemplate.afterPropertiesSet(); + if (!amqpTemplateExplicitlySet) { + ((RabbitTemplate) this.amqpTemplate).afterPropertiesSet(); + } super.onInit(); } diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd index 06a3b2baf9..ce1a8b9447 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd @@ -229,6 +229,31 @@ ]]> + + + + + + + + + The AmqpTemplate bean reference to be used for sending replies. + Defaults to `RabbitTemplate` based on the provided `ConnectionFactory`. + + + + + + + The 'defaultReplyTo' address with the form '(exchange)/(routingKey)' + (or '(queueName)' - in which case the default exchange will be used + with the queue name as the routing key) + if the request message doesn't have 'replyTo' property. + If this property isn't specified too, the gateway relies on + the `AmqpTemplate` configuration. + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml index c8593fcb3c..b7da686bf8 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests-context.xml @@ -20,16 +20,22 @@ + + + + - + - + - + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java index c0e0ce2930..1158c97fe2 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -29,6 +29,7 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.springframework.amqp.core.Address; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; @@ -88,6 +89,9 @@ public class AmqpInboundGatewayParserTests { assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(gateway, "autoStartup")); assertEquals(123, TestUtils.getPropertyValue(gateway, "phase")); assertFalse(TestUtils.getPropertyValue(gateway, "messageListenerContainer.missingQueuesFatal", Boolean.class)); + Object amqpTemplate = context.getBean("amqpTemplate"); + assertSame(amqpTemplate, TestUtils.getPropertyValue(gateway, "amqpTemplate")); + assertEquals(new Address("fooExchange/barRoutingKey"), TestUtils.getPropertyValue(gateway, "defaultReplyTo")); } @SuppressWarnings("rawtypes") diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index d0684f43d8..b067de7d3e 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 the original author or authors. + * Copyright 2013-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -44,7 +44,6 @@ import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.channel.DirectChannel; @@ -53,7 +52,6 @@ import org.springframework.integration.json.JsonToObjectTransformer; import org.springframework.integration.json.ObjectToJsonTransformer; import org.springframework.integration.mapping.support.JsonHeaders; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.integration.transformer.Transformer; import org.springframework.messaging.Message; @@ -186,15 +184,7 @@ public class InboundEndpointTests { } })); - AmqpInboundGateway gateway = new AmqpInboundGateway(container); - gateway.setMessageConverter(new Jackson2JsonMessageConverter()); - - gateway.setRequestChannel(channel); - gateway.setBeanFactory(mock(BeanFactory.class)); - gateway.afterPropertiesSet(); - - RabbitTemplate rabbitTemplate = Mockito.spy(TestUtils.getPropertyValue(gateway, "amqpTemplate", - RabbitTemplate.class)); + RabbitTemplate rabbitTemplate = Mockito.mock(RabbitTemplate.class); Mockito.doAnswer(new Answer() { @@ -215,13 +205,17 @@ public class InboundEndpointTests { }).when(rabbitTemplate).send(Mockito.anyString(), Mockito.anyString(), Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class)); - DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(gateway); - directFieldAccessor.setPropertyValue("amqpTemplate", rabbitTemplate); + AmqpInboundGateway gateway = new AmqpInboundGateway(container, rabbitTemplate); + gateway.setMessageConverter(new Jackson2JsonMessageConverter()); + gateway.setRequestChannel(channel); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setDefaultReplyTo("foo"); + gateway.afterPropertiesSet(); + Object payload = new Foo("bar1"); MessageProperties amqpMessageProperties = new MessageProperties(); - amqpMessageProperties.setReplyTo("test"); amqpMessageProperties.setDeliveryTag(123L); org.springframework.amqp.core.Message amqpMessage = new Jackson2JsonMessageConverter().toMessage(payload, amqpMessageProperties); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index c6bebb8e7d..45dc99d0bd 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -227,7 +227,9 @@ The inbound gateway supports all the attributes on the inbound channel adapter ( mapped-request-headers="" <4> mapped-reply-headers="" <5> reply-channel="myReplyChannel" <6> - reply-timeout="1000" /> <7> + reply-timeout="1000" <7> + amqp-template="" <8> + default-reply-to="" /> <9> ---- @@ -268,6 +270,17 @@ _Optional_. If not specified this property will default to "1000" (1 second). Only applies if the container thread hands off to another thread before the reply is sent. +<8> The customized `AmqpTemplate` bean reference to have more control for the reply messages to send or you can provide +an alternative implementation to the `RabbitTemplate`. + +<9> The `replyTo` `org.springframework.amqp.core.Address` to be used when the `requestMessage` doesn't have `replyTo` +property. +If this option isn't specified, no `amqp-template` is provided, and no `replyTo` property exists in the request message, +an `IllegalStateException` is thrown because the reply can't be routed. +If this option isn't specified, and an external `amqp-template` is provided, no exception will be thrown. +You __must__ either specify this option, or configure a default `exchange` and `routingKey` on that template, +if you anticipate cases when no `replyTo` property exists in the request message. + See the note in <> about configuring the `listener-container` attribute. [[amqp-inbound-ack]] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 285796d29d..f9aa3f2561 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -252,10 +252,18 @@ See <> for more information. The `` now supports `confirm-correlation-expression` and `confirm-(n)ack-channel` attributes with similar purpose as for ``. +===== Correlation Data + For both the outbound channel adapter and gateway, if the correlation data is a `Message`, it will be the basis of the message on the ack/nack channel, with the additional header(s) added. Previously, any correlation data (including `Message`) was returned as the payload of the ack/nack message. +===== The Inbound Gateway properties + +The `` now exposes the `amqp-template` attribute to allow more control over an external bean +for the reply `RabbitTemplate` or even provide your own `AmqpTemplate` implementation. +In addition the `default-reply-to` is exposed to be used if request message doesn't have `replyTo` property. + See <> for more information. [[x4.2-xpath-splitter]]