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