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

Merge pull request #640 from artembilan/INT-2773

Browse Source
This commit is contained in:
Gary Russell
2012-10-11 18:37:32 -04:00
parent ef2549cf22 eb1500cb7e
commit dc63007352
5 changed files with 134 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -28,7 +28,7 @@ import org.w3c.dom.Element;
/**
* Parser for the AMQP 'outbound-channel-adapter' element.
*
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
@@ -44,9 +44,9 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
amqpTemplateRef = "amqpTemplate";
}
builder.addConstructorArgReference(amqpTemplateRef);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name", true);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key", true);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);

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

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 the original author or authors.
*
* Copyright 2002-2012 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.
@@ -39,10 +39,11 @@ import org.springframework.util.Assert;
/**
* Adapter that converts and sends Messages to an AMQP Exchange.
*
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
@@ -55,9 +56,9 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
private volatile boolean expectReply;
private volatile String exchangeName = "";
private volatile String exchangeName;
private volatile String routingKey = "";
private volatile String routingKey;
private volatile String exchangeNameExpression;
@@ -82,15 +83,15 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
@Override
protected void onInit() {
super.onInit();
Assert.state(exchangeNameExpression == null || "".equals(exchangeName),
Assert.state(exchangeNameExpression == null || exchangeName == null,
"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
Assert.state(this.confirmCorrelationExpression != null ? !this.expectReply : true,
Assert.state(this.confirmCorrelationExpression == null || !this.expectReply,
"Confirm correlation expression does not apply to a gateway");
if (exchangeNameExpression != null) {
Expression expression = expressionParser.parseExpression(this.exchangeNameExpression);
this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
}
Assert.state(routingKeyExpression == null || "".equals(routingKey),
Assert.state(routingKeyExpression == null || routingKey == null,
"Either a routingKey or a routingKeyExpression can be provided, but not both");
if (routingKeyExpression != null) {
Expression expression = expressionParser.parseExpression(this.routingKeyExpression);
@@ -104,20 +105,22 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
if (this.returnChannel != null) {
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for publisher returns");
((RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
( (RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
}
}
public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
Assert.notNull(amqpTemplate, "AmqpTemplate must not be null");
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;
}
@@ -126,6 +129,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
public void setRoutingKey(String routingKey) {
Assert.notNull(routingKey, "routingKey must not be null");
this.routingKey = routingKey;
}

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

@@ -24,7 +24,7 @@
<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"/>
<int:channel id="requestChannel"/>
<int:chain id="chainWithRabbitOutbound" input-channel="amqpOutboundChannelAdapterWithinChain">
@@ -54,4 +54,14 @@
<int:queue/>
</int:channel>
<amqp:outbound-channel-adapter id="withDefaultAmqpTemplateExchangeAndRoutingKey"/>
<rabbit:template id="amqpTemplateWithSuppliedExchangeAndRoutingKey" connection-factory="connectionFactory"
exchange="default.test.exchange" routing-key="default.routing.key"/>
<amqp:outbound-channel-adapter id="toRabbitOnlyWithTemplateChannel" amqp-template="amqpTemplateWithSuppliedExchangeAndRoutingKey"/>
<amqp:outbound-channel-adapter id="overrideTemplateAttributesToEmpty" amqp-template="amqpTemplateWithSuppliedExchangeAndRoutingKey"
exchange-name="" routing-key=""/>
</beans>

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

@@ -25,6 +25,7 @@ import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.List;
@@ -117,12 +118,12 @@ public class AmqpOutboundChannelAdapterParserTests {
assertEquals("foobar", properties.getHeaders().get("foobar"));
assertNull(properties.getHeaders().get("bar"));
return null;
}})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class));
}
})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("requestChannel", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").setHeader("bar", "bar").setHeader("foobar", "foobar").build();
requestChannel.send(message);
@@ -183,12 +184,12 @@ public class AmqpOutboundChannelAdapterParserTests {
org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2];
assertEquals("hello", new String(amqpReplyMessage.getBody()));
return null;
}})
}
})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class),
Mockito.any(CorrelationData.class));
Mockito.any(CorrelationData.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("amqpOutboundChannelAdapterWithinChain", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").build();
requestChannel.send(message);
@@ -210,7 +211,7 @@ public class AmqpOutboundChannelAdapterParserTests {
Message<?> message = MessageBuilder.withPayload("hello").build();
requestChannel.send(message);
PollableChannel returnChannel = context.getBean("returnChannel", PollableChannel.class);
RabbitTemplate template = context.getBean(RabbitTemplate.class);
RabbitTemplate template = context.getBean("amqpTemplate", RabbitTemplate.class);
Map<String, Object> headers = new HashMap<String, Object>();
headers.put(PublisherCallbackChannel.RETURN_CORRELATION, template.getUUID());
BasicProperties properties = mock(BasicProperties.class);
@@ -239,6 +240,54 @@ public class AmqpOutboundChannelAdapterParserTests {
}
}
@Test
public void testInt2773UseDefaultAmqpTemplateExchangeAndRoutingLey() throws IOException {
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(mockConnection);
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
MessageChannel requestChannel = context.getBean("toRabbitOnlyWithTemplateChannel", MessageChannel.class);
requestChannel.send(MessageBuilder.withPayload("test").build());
Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq("default.test.exchange"), Mockito.eq("default.routing.key"),
Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
}
@Test
public void testInt2773WithDefaultAmqpTemplateExchangeAndRoutingLey() throws IOException {
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(mockConnection);
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
MessageChannel requestChannel = context.getBean("withDefaultAmqpTemplateExchangeAndRoutingKey", MessageChannel.class);
requestChannel.send(MessageBuilder.withPayload("test").build());
Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq(""), Mockito.eq(""),
Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
}
@Test
public void testInt2773WithOverrideToDefaultAmqpTemplateExchangeAndRoutingLey() throws IOException {
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(mockConnection);
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
MessageChannel requestChannel = context.getBean("overrideTemplateAttributesToEmpty", MessageChannel.class);
requestChannel.send(MessageBuilder.withPayload("test").build());
Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq(""), Mockito.eq(""),
Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@Override

55
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -17,10 +17,6 @@ import static org.springframework.beans.factory.xml.AbstractBeanDefinitionParser
import java.util.List;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.RuntimeBeanReference;
@@ -41,6 +37,9 @@ import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
/**
* Shared utility methods for integration namespace parsers.
@@ -90,10 +89,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName,
String propertyName) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyValue(propertyName, new TypedStringValue(attributeValue));
}
setValueIfAttributeDefined(builder, element, attributeName, propertyName, false);
}
/**
@@ -111,8 +107,49 @@ public abstract class IntegrationNamespaceUtils {
* @param attributeName - the name of the attribute whose value will be set on the property
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName) {
setValueIfAttributeDefined(builder, element, attributeName, false);
}
/**
* Configures the provided bean definition builder with a property value corresponding to the attribute whose name
* is provided if that attribute is defined in the given element.
*
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used to populate the property
* @param propertyName the name of the property to be populated
* @param emptyStringAllowed - if true, the value is set, even if an empty String (""); if false, an empty
* String is treated as if the attribute wasn't provided.
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName,
String propertyName, boolean emptyStringAllowed) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue) || (emptyStringAllowed && element.hasAttribute(attributeName))) {
builder.addPropertyValue(propertyName, new TypedStringValue(attributeValue));
}
}
/**
* Configures the provided bean definition builder with a property value corresponding to the attribute whose name
* is provided if that attribute is defined in the given element.
*
* <p>
* The property name will be the camel-case equivalent of the lower case hyphen separated attribute (e.g. the
* "foo-bar" attribute would match the "fooBar" property).
*
* @see Conventions#attributeNameToPropertyName(String)
*
* @param builder the bean definition builder to be configured
* @param element - the XML element where the attribute should be defined
* @param attributeName - the name of the attribute whose value will be set on the property
* @param emptyStringAllowed - if true, the value is set, even if an empty String (""); if false, an empty
* String is treated as if the attribute wasn't provided.
*
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName,
boolean emptyStringAllowed) {
setValueIfAttributeDefined(builder, element, attributeName,
Conventions.attributeNameToPropertyName(attributeName));
Conventions.attributeNameToPropertyName(attributeName), emptyStringAllowed);
}
/**