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

INT-3807: AmqpInboundGateway Improvements

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

Addressing PR comments

INT-3807: Polishing
This commit is contained in:
Artem Bilan
2015-08-20 11:44:32 -04:00
committed by Gary Russell
parent 7464168845
commit 02565d0401
9 changed files with 176 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

9
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java
View File

@@ -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");
}
}

132
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -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
* <pre class="code">
* (exchange)/(routingKey)
* </pre>
* or
* <pre class="code">
* (queueName)
* </pre>
* 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();
}

25
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd
View File

@@ -229,6 +229,31 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="amqp-template" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The AmqpTemplate bean reference to be used for sending replies.
Defaults to `RabbitTemplate` based on the provided `ConnectionFactory`.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="default-reply-to">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

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

@@ -20,16 +20,22 @@
<bean id="testConverter" class="org.springframework.integration.amqp.config.AmqpInboundGatewayParserTests$TestConverter"/>
<bean id="amqpTemplate" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.core.AmqpTemplate"/>
</bean>
<si-amqp:inbound-gateway id="autoStartFalseGateway" request-channel="requests" queue-names="test"
connection-factory="rabbitConnectionFactory" message-converter="testConverter"
missing-queues-fatal="false"
amqp-template="amqpTemplate"
default-reply-to="fooExchange/barRoutingKey"
auto-startup="false" phase="123"/>
<si-amqp:inbound-gateway id="withHeaderMapper" request-channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers="foo*, STANDARD_REQUEST_HEADERS"
mapped-reply-headers="bar*"/>
<int:channel id="requestChannel"/>
</beans>

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

@@ -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")

24
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -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<Object>() {
@@ -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);

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

@@ -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 <<amqp-inbound-channel-adapter>> about configuring the `listener-container` attribute.
[[amqp-inbound-ack]]

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

@@ -252,10 +252,18 @@ See <<conditional-pollers>> for more information.
The `<int-amqp:outbound-gateway>` now supports `confirm-correlation-expression` and `confirm-(n)ack-channel`
attributes with similar purpose as for `<int-amqp:outbound-channel-adapter>`.
===== 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 `<int-amqp:inbound-gateway>` 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 <<amqp>> for more information.
[[x4.2-xpath-splitter]]