From eb1500cb7e024930b82f8e64d002afe6719974ee Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 2 Oct 2012 16:12:49 +0300 Subject: [PATCH] INT-2773: Fix `AmqpOutboundEndpoint`'s properties * make `AmqpOutboundEndpoint` 'exchangeName' & 'routingKey' **null** by default * some polishing for `AmqpOutboundEndpoint` * verification tests for parameters of `com.rabbitmq.client.Channel#basicPublish` JIRA: https://jira.springsource.org/browse/INT-2773 INT-2773 Polishing - Allow Override to "" Previously, there was no way to revert to the previous behavior by specifying an empty string for the attribute(s). Add test to verify setting attributes to "" overrides template. --- .../AmqpOutboundChannelAdapterParser.java | 6 +- .../amqp/outbound/AmqpOutboundEndpoint.java | 30 +++++---- ...boundChannelAdapterParserTests-context.xml | 12 +++- ...AmqpOutboundChannelAdapterParserTests.java | 65 ++++++++++++++++--- .../config/xml/IntegrationNamespaceUtils.java | 55 +++++++++++++--- 5 files changed, 134 insertions(+), 34 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java index 63ccd94067..601d9822ea 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java @@ -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); 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 bb912ea81c..cd1a074390 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 @@ -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(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; } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml index ceaa5e15f4..563c869c6c 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml @@ -24,7 +24,7 @@ - + @@ -54,4 +54,14 @@ + + + + + + + + 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 d1fd4663c6..dee100ca9e 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 @@ -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 headers = new HashMap(); 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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java index 361cad56bd..1f0c3b15cd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java @@ -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. + * + *

+ * 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); } /**