diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java index 756f0fc2..b8f7fc55 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java @@ -33,6 +33,8 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { private static final String PORT_ATTRIBUTE = "port"; + private static final String ADDRESSES = "addresses"; + private static final String VIRTUAL_HOST_ATTRIBUTE = "virtual-host"; private static final String USER_ATTRIBUTE = "username"; @@ -58,7 +60,11 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - + if (element.hasAttribute(ADDRESSES) && + (element.hasAttribute(HOST_ATTRIBUTE) || element.hasAttribute(PORT_ATTRIBUTE))) { + parserContext.getReaderContext().error("If the 'addresses' attribute is provided, a connection " + + "factory can not have 'host' or 'port' attributes.", element); + } NamespaceUtils.addConstructorArgParentRefIfAttributeDefined(builder, element, CONNECTION_FACTORY_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, CHANNEL_CACHE_SIZE_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, HOST_ATTRIBUTE); @@ -67,6 +73,7 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setValueIfAttributeDefined(builder, element, PASSWORD_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, VIRTUAL_HOST_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE); + NamespaceUtils.setValueIfAttributeDefined(builder, element, ADDRESSES); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java index 022d8b28..7394cf46 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java @@ -15,8 +15,6 @@ package org.springframework.amqp.rabbit.config; import java.util.List; -import org.springframework.amqp.core.AcknowledgeMode; -import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.config.TypedStringValue; @@ -38,12 +36,6 @@ import org.w3c.dom.NodeList; */ class ListenerContainerParser implements BeanDefinitionParser { - private static final String CONNECTION_FACTORY_ATTRIBUTE = "connection-factory"; - - private static final String TASK_EXECUTOR_ATTRIBUTE = "task-executor"; - - private static final String ERROR_HANDLER_ATTRIBUTE = "error-handler"; - private static final String LISTENER_ELEMENT = "listener"; private static final String ID_ATTRIBUTE = "id"; @@ -62,30 +54,6 @@ class ListenerContainerParser implements BeanDefinitionParser { private static final String RESPONSE_ROUTING_KEY_ATTRIBUTE = "response-routing-key"; - private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge"; - - private static final String ACKNOWLEDGE_AUTO = "auto"; - - private static final String ACKNOWLEDGE_MANUAL = "manual"; - - private static final String ACKNOWLEDGE_NONE = "none"; - - private static final String TRANSACTION_MANAGER_ATTRIBUTE = "transaction-manager"; - - private static final String CONCURRENCY_ATTRIBUTE = "concurrency"; - - private static final String PREFETCH_ATTRIBUTE = "prefetch"; - - private static final String TRANSACTION_SIZE_ATTRIBUTE = "transaction-size"; - - private static final String PHASE_ATTRIBUTE = "phase"; - - private static final String AUTO_STARTUP_ATTRIBUTE = "auto-startup"; - - private static final String ADVICE_CHAIN_ATTRIBUTE = "advice-chain"; - - private static final String REQUEUE_REJECTED_ATTRIBUTE = "requeue-rejected"; - public BeanDefinition parse(Element element, ParserContext parserContext) { CompositeComponentDefinition compositeDef = new CompositeComponentDefinition(element.getTagName(), parserContext.extractSource(element)); @@ -137,7 +105,7 @@ class ListenerContainerParser implements BeanDefinitionParser { } } - BeanDefinition containerDef = parseContainer(listenerEle, containerEle, parserContext); + BeanDefinition containerDef = RabbitNamespaceUtils.parseContainer(containerEle, parserContext); if (listenerEle.hasAttribute(RESPONSE_EXCHANGE_ATTRIBUTE)) { String responseExchange = listenerEle.getAttribute(RESPONSE_EXCHANGE_ATTRIBUTE); @@ -191,102 +159,4 @@ class ListenerContainerParser implements BeanDefinitionParser { // Register the listener and fire event parserContext.registerBeanComponent(new BeanComponentDefinition(containerDef, containerBeanName)); } - - private BeanDefinition parseContainer(Element listenerEle, Element containerEle, ParserContext parserContext) { - RootBeanDefinition containerDef = new RootBeanDefinition(SimpleMessageListenerContainer.class); - containerDef.setSource(parserContext.extractSource(containerEle)); - - String connectionFactoryBeanName = "rabbitConnectionFactory"; - if (containerEle.hasAttribute(CONNECTION_FACTORY_ATTRIBUTE)) { - connectionFactoryBeanName = containerEle.getAttribute(CONNECTION_FACTORY_ATTRIBUTE); - if (!StringUtils.hasText(connectionFactoryBeanName)) { - parserContext.getReaderContext().error( - "Listener container 'connection-factory' attribute contains empty value.", containerEle); - } - } - if (StringUtils.hasText(connectionFactoryBeanName)) { - containerDef.getPropertyValues().add("connectionFactory", - new RuntimeBeanReference(connectionFactoryBeanName)); - } - - String taskExecutorBeanName = containerEle.getAttribute(TASK_EXECUTOR_ATTRIBUTE); - if (StringUtils.hasText(taskExecutorBeanName)) { - containerDef.getPropertyValues().add("taskExecutor", new RuntimeBeanReference(taskExecutorBeanName)); - } - - String errorHandlerBeanName = containerEle.getAttribute(ERROR_HANDLER_ATTRIBUTE); - if (StringUtils.hasText(errorHandlerBeanName)) { - containerDef.getPropertyValues().add("errorHandler", new RuntimeBeanReference(errorHandlerBeanName)); - } - - AcknowledgeMode acknowledgeMode = parseAcknowledgeMode(containerEle, parserContext); - if (acknowledgeMode != null) { - containerDef.getPropertyValues().add("acknowledgeMode", acknowledgeMode); - } - - String transactionManagerBeanName = containerEle.getAttribute(TRANSACTION_MANAGER_ATTRIBUTE); - if (StringUtils.hasText(transactionManagerBeanName)) { - containerDef.getPropertyValues().add("transactionManager", - new RuntimeBeanReference(transactionManagerBeanName)); - } - - String concurrency = containerEle.getAttribute(CONCURRENCY_ATTRIBUTE); - if (StringUtils.hasText(concurrency)) { - containerDef.getPropertyValues().add("concurrentConsumers", new TypedStringValue(concurrency)); - } - - String prefetch = containerEle.getAttribute(PREFETCH_ATTRIBUTE); - if (StringUtils.hasText(prefetch)) { - containerDef.getPropertyValues().add("prefetchCount", new TypedStringValue(prefetch)); - } - - String transactionSize = containerEle.getAttribute(TRANSACTION_SIZE_ATTRIBUTE); - if (StringUtils.hasText(transactionSize)) { - containerDef.getPropertyValues().add("txSize", new TypedStringValue(transactionSize)); - } - - String requeueRejected = containerEle.getAttribute(REQUEUE_REJECTED_ATTRIBUTE); - if (StringUtils.hasText(requeueRejected)) { - containerDef.getPropertyValues().add("defaultRequeueRejected", new TypedStringValue(requeueRejected)); - } - - String phase = containerEle.getAttribute(PHASE_ATTRIBUTE); - if (StringUtils.hasText(phase)) { - containerDef.getPropertyValues().add("phase", phase); - } - - String autoStartup = containerEle.getAttribute(AUTO_STARTUP_ATTRIBUTE); - if (StringUtils.hasText(autoStartup)) { - containerDef.getPropertyValues().add("autoStartup", new TypedStringValue(autoStartup)); - } - - String adviceChain = containerEle.getAttribute(ADVICE_CHAIN_ATTRIBUTE); - if (StringUtils.hasText(adviceChain)) { - containerDef.getPropertyValues().add("adviceChain", new RuntimeBeanReference(adviceChain)); - } - - return containerDef; - } - - private AcknowledgeMode parseAcknowledgeMode(Element ele, ParserContext parserContext) { - AcknowledgeMode acknowledgeMode = null; - String acknowledge = ele.getAttribute(ACKNOWLEDGE_ATTRIBUTE); - if (StringUtils.hasText(acknowledge)) { - if (ACKNOWLEDGE_AUTO.equals(acknowledge)) { - acknowledgeMode = AcknowledgeMode.AUTO; - } else if (ACKNOWLEDGE_MANUAL.equals(acknowledge)) { - acknowledgeMode = AcknowledgeMode.MANUAL; - } else if (ACKNOWLEDGE_NONE.equals(acknowledge)) { - acknowledgeMode = AcknowledgeMode.NONE; - } else { - parserContext.getReaderContext().error( - "Invalid listener container 'acknowledge' setting [" + acknowledge - + "]: only \"auto\", \"manual\", and \"none\" supported.", ele); - } - return acknowledgeMode; - } else { - return null; - } - } - } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueArgumentsParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueArgumentsParser.java new file mode 100644 index 00000000..e1eaaeed --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueArgumentsParser.java @@ -0,0 +1,46 @@ +/* + * 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. + */ +package org.springframework.amqp.rabbit.config; + +import java.util.Map; + +import org.springframework.beans.factory.config.MapFactoryBean; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.w3c.dom.Element; + +/** + * @author Gary Russell + * @since 1.0.1 + * + */ +class QueueArgumentsParser extends AbstractSingleBeanDefinitionParser { + + @Override + protected void doParse(Element element, ParserContext parserContext, + BeanDefinitionBuilder builder) { + Map map = parserContext.getDelegate().parseMapElement(element, + builder.getRawBeanDefinition()); + builder.addPropertyValue("sourceMap", map); + } + + @Override + protected String getBeanClassName(Element element) { + return MapFactoryBean.class.getName(); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java index 8cbeb0d8..e3ab0dba 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java @@ -20,6 +20,7 @@ import org.springframework.amqp.core.Queue; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; import org.w3c.dom.Element; @@ -29,7 +30,7 @@ import org.w3c.dom.Element; */ public class QueueParser extends AbstractSingleBeanDefinitionParser { - private static final String ARGUMENTS_ELEMENT = "queue-arguments"; + private static final String ARGUMENTS = "queue-arguments"; // element OR attribute private static final String DURABLE_ATTRIBUTE = "durable"; private static final String EXCLUSIVE_ATTRIBUTE = "exclusive"; private static final String AUTO_DELETE_ATTRIBUTE = "auto-delete"; @@ -79,13 +80,25 @@ public class QueueParser extends AbstractSingleBeanDefinitionParser { } - Element argumentsElement = DomUtils.getChildElementByTagName(element, ARGUMENTS_ELEMENT); + String queueArguments = element.getAttribute(ARGUMENTS); + Element argumentsElement = DomUtils.getChildElementByTagName(element, ARGUMENTS); + if (argumentsElement != null) { + if (StringUtils.hasText(queueArguments)) { + parserContext + .getReaderContext() + .error("Queue may have either a queue-attributes attribute or element, but not both", + element); + } Map map = parserContext.getDelegate().parseMapElement(argumentsElement, builder.getRawBeanDefinition()); builder.addConstructorArgValue(map); } + if (StringUtils.hasText(queueArguments)) { + builder.addConstructorArgReference(queueArguments); + } + } private boolean attributeHasIllegalOverride(Element element, String name, String allowed) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java index b1060aaf..1a660aa0 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java @@ -23,6 +23,7 @@ import org.springframework.beans.factory.xml.NamespaceHandlerSupport; * * @author Mark Pollack * @author Mark Fisher + * @author Gary Russell * @since 1.0 */ public class RabbitNamespaceHandler extends NamespaceHandlerSupport { @@ -37,6 +38,7 @@ public class RabbitNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("admin", new AdminParser()); registerBeanDefinitionParser("connection-factory", new ConnectionFactoryParser()); registerBeanDefinitionParser("template", new TemplateParser()); + registerBeanDefinitionParser("queue-arguments", new QueueArgumentsParser()); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java new file mode 100644 index 00000000..cc7324c4 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java @@ -0,0 +1,162 @@ +/* + * 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. + */ +package org.springframework.amqp.rabbit.config; + +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.config.TypedStringValue; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + +/** + * @author Gary Russell + * @since 1.0.1 + * + */ +public class RabbitNamespaceUtils { + + private static final String CONNECTION_FACTORY_ATTRIBUTE = "connection-factory"; + + private static final String TASK_EXECUTOR_ATTRIBUTE = "task-executor"; + + private static final String ERROR_HANDLER_ATTRIBUTE = "error-handler"; + + private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge"; + + private static final String ACKNOWLEDGE_AUTO = "auto"; + + private static final String ACKNOWLEDGE_MANUAL = "manual"; + + private static final String ACKNOWLEDGE_NONE = "none"; + + private static final String TRANSACTION_MANAGER_ATTRIBUTE = "transaction-manager"; + + private static final String CONCURRENCY_ATTRIBUTE = "concurrency"; + + private static final String PREFETCH_ATTRIBUTE = "prefetch"; + + private static final String TRANSACTION_SIZE_ATTRIBUTE = "transaction-size"; + + private static final String PHASE_ATTRIBUTE = "phase"; + + private static final String AUTO_STARTUP_ATTRIBUTE = "auto-startup"; + + private static final String ADVICE_CHAIN_ATTRIBUTE = "advice-chain"; + + private static final String REQUEUE_REJECTED_ATTRIBUTE = "requeue-rejected"; + + public static BeanDefinition parseContainer(Element containerEle, ParserContext parserContext) { + RootBeanDefinition containerDef = new RootBeanDefinition(SimpleMessageListenerContainer.class); + containerDef.setSource(parserContext.extractSource(containerEle)); + + String connectionFactoryBeanName = "rabbitConnectionFactory"; + if (containerEle.hasAttribute(CONNECTION_FACTORY_ATTRIBUTE)) { + connectionFactoryBeanName = containerEle.getAttribute(CONNECTION_FACTORY_ATTRIBUTE); + if (!StringUtils.hasText(connectionFactoryBeanName)) { + parserContext.getReaderContext().error( + "Listener container 'connection-factory' attribute contains empty value.", containerEle); + } + } + if (StringUtils.hasText(connectionFactoryBeanName)) { + containerDef.getPropertyValues().add("connectionFactory", + new RuntimeBeanReference(connectionFactoryBeanName)); + } + + String taskExecutorBeanName = containerEle.getAttribute(TASK_EXECUTOR_ATTRIBUTE); + if (StringUtils.hasText(taskExecutorBeanName)) { + containerDef.getPropertyValues().add("taskExecutor", new RuntimeBeanReference(taskExecutorBeanName)); + } + + String errorHandlerBeanName = containerEle.getAttribute(ERROR_HANDLER_ATTRIBUTE); + if (StringUtils.hasText(errorHandlerBeanName)) { + containerDef.getPropertyValues().add("errorHandler", new RuntimeBeanReference(errorHandlerBeanName)); + } + + AcknowledgeMode acknowledgeMode = parseAcknowledgeMode(containerEle, parserContext); + if (acknowledgeMode != null) { + containerDef.getPropertyValues().add("acknowledgeMode", acknowledgeMode); + } + + String transactionManagerBeanName = containerEle.getAttribute(TRANSACTION_MANAGER_ATTRIBUTE); + if (StringUtils.hasText(transactionManagerBeanName)) { + containerDef.getPropertyValues().add("transactionManager", + new RuntimeBeanReference(transactionManagerBeanName)); + } + + String concurrency = containerEle.getAttribute(CONCURRENCY_ATTRIBUTE); + if (StringUtils.hasText(concurrency)) { + containerDef.getPropertyValues().add("concurrentConsumers", new TypedStringValue(concurrency)); + } + + String prefetch = containerEle.getAttribute(PREFETCH_ATTRIBUTE); + if (StringUtils.hasText(prefetch)) { + containerDef.getPropertyValues().add("prefetchCount", new TypedStringValue(prefetch)); + } + + String transactionSize = containerEle.getAttribute(TRANSACTION_SIZE_ATTRIBUTE); + if (StringUtils.hasText(transactionSize)) { + containerDef.getPropertyValues().add("txSize", new TypedStringValue(transactionSize)); + } + + String requeueRejected = containerEle.getAttribute(REQUEUE_REJECTED_ATTRIBUTE); + if (StringUtils.hasText(requeueRejected)) { + containerDef.getPropertyValues().add("defaultRequeueRejected", new TypedStringValue(requeueRejected)); + } + + String phase = containerEle.getAttribute(PHASE_ATTRIBUTE); + if (StringUtils.hasText(phase)) { + containerDef.getPropertyValues().add("phase", phase); + } + + String autoStartup = containerEle.getAttribute(AUTO_STARTUP_ATTRIBUTE); + if (StringUtils.hasText(autoStartup)) { + containerDef.getPropertyValues().add("autoStartup", new TypedStringValue(autoStartup)); + } + + String adviceChain = containerEle.getAttribute(ADVICE_CHAIN_ATTRIBUTE); + if (StringUtils.hasText(adviceChain)) { + containerDef.getPropertyValues().add("adviceChain", new RuntimeBeanReference(adviceChain)); + } + + return containerDef; + } + + private static AcknowledgeMode parseAcknowledgeMode(Element ele, ParserContext parserContext) { + AcknowledgeMode acknowledgeMode = null; + String acknowledge = ele.getAttribute(ACKNOWLEDGE_ATTRIBUTE); + if (StringUtils.hasText(acknowledge)) { + if (ACKNOWLEDGE_AUTO.equals(acknowledge)) { + acknowledgeMode = AcknowledgeMode.AUTO; + } else if (ACKNOWLEDGE_MANUAL.equals(acknowledge)) { + acknowledgeMode = AcknowledgeMode.MANUAL; + } else if (ACKNOWLEDGE_NONE.equals(acknowledge)) { + acknowledgeMode = AcknowledgeMode.NONE; + } else { + parserContext.getReaderContext().error( + "Invalid listener container 'acknowledge' setting [" + acknowledge + + "]: only \"auto\", \"manual\", and \"none\" supported.", ele); + } + return acknowledgeMode; + } else { + return null; + } + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java index 5baaee6e..9a85f235 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2010-2011 the original author or authors. + * Copyright 2010-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 @@ -14,14 +14,19 @@ package org.springframework.amqp.rabbit.config; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; import org.w3c.dom.Element; +import org.w3c.dom.Node; +import org.w3c.dom.NodeList; /** * @author Dave Syer + * @author Gary Russell */ class TemplateParser extends AbstractSingleBeanDefinitionParser { @@ -41,6 +46,10 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { private static final String CHANNEL_TRANSACTED_ATTRIBUTE = "channel-transacted"; + private static final String REPLY_QUEUE_ATTRIBUTE = "reply-queue"; + + private static final String LISTENER_ELEMENT = "reply-listener"; + @Override protected Class getBeanClass(Element element) { return RabbitTemplate.class; @@ -53,7 +62,7 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { @Override protected boolean shouldGenerateIdAsFallback() { - return true; + return false; } @Override @@ -77,7 +86,65 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setValueIfAttributeDefined(builder, element, REPLY_TIMEOUT_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, ENCODING_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_CONVERTER_ATTRIBUTE); + NamespaceUtils.setReferenceIfAttributeDefined(builder, element, REPLY_QUEUE_ATTRIBUTE); + BeanDefinition replyContainer = null; + Element childElement = getChildElement(element, parserContext); + if (childElement != null) { + replyContainer = parseListener(childElement, element, + parserContext); + if (replyContainer != null) { + replyContainer.getPropertyValues().add("messageListener", + new RuntimeBeanReference(element.getAttribute(ID_ATTRIBUTE))); + String replyContainerName = element.getAttribute(ID_ATTRIBUTE) + ".replyListener"; + parserContext.getRegistry().registerBeanDefinition(replyContainerName, replyContainer); + } + } + if (replyContainer == null && element.hasAttribute(REPLY_QUEUE_ATTRIBUTE)) { + parserContext.getReaderContext().error( + "For template '" + element.getAttribute(ID_ATTRIBUTE) + + "', when specifying a reply-queue, " + + "a element is required", + element); + } + else if (replyContainer != null && !element.hasAttribute(REPLY_QUEUE_ATTRIBUTE)) { + parserContext.getReaderContext().error( + "For template '" + element.getAttribute(ID_ATTRIBUTE) + + "', a element is not allowed if no " + + "'reply-queue' is supplied", + element); + } + } + + private Element getChildElement(Element element, + ParserContext parserContext) { + Element childElement = null; + NodeList childNodes = element.getChildNodes(); + for (int i = 0; i < childNodes.getLength(); i++) { + Node child = childNodes.item(i); + if (child.getNodeType() == Node.ELEMENT_NODE) { + String localName = parserContext.getDelegate().getLocalName(child); + if (LISTENER_ELEMENT.equals(localName)) { + childElement = (Element) child; + } + } + } + return childElement; + } + + private BeanDefinition parseListener(Element childElement, Element element, + ParserContext parserContext) { + if (getChildElement(childElement, parserContext) != null) { + parserContext.getReaderContext().error(" is not allowed any child elements.", element); + } + BeanDefinition replyContainer = RabbitNamespaceUtils.parseContainer(childElement, parserContext); + if (replyContainer != null) { + replyContainer.getPropertyValues().add( + "connectionFactory", + new RuntimeBeanReference(element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE))); + } + replyContainer.getPropertyValues().add("queues", element.getAttribute(REPLY_QUEUE_ATTRIBUTE)); + return replyContainer; } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index 65058fa8..89eaf18e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -25,6 +25,8 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.Assert; +import com.rabbitmq.client.Address; + /** * @author Dave Syer * @author Gary Russell @@ -42,6 +44,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di private volatile ExecutorService executorService; + private volatile Address[] addresses; + /** * Create a new SingleConnectionFactory for the given target ConnectionFactory. * @param rabbitConnectionFactory the target ConnectionFactory @@ -83,6 +87,17 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di return this.rabbitConnectionFactory.getPort(); } + /** + * Set addresses for clustering. + * @param addresses list of addresses with form "host[:port],..." + */ + public void setAddresses(String addresses) { + Address[] addressArray = Address.parseAddresses(addresses); + if (addressArray.length > 0) { + this.addresses = addressArray; + } + } + /** * A composite connection listener to be used by subclasses when creating and closing connections. * @@ -138,7 +153,12 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di final protected Connection createBareConnection() { try { - return new SimpleConnection(this.rabbitConnectionFactory.newConnection(this.executorService)); + if (this.addresses != null) { + return new SimpleConnection(this.rabbitConnectionFactory.newConnection(this.executorService, this.addresses)); + } + else { + return new SimpleConnection(this.rabbitConnectionFactory.newConnection(this.executorService)); + } } catch (IOException e) { throw RabbitUtils.convertRabbitAccessException(e); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 0f431d82..705d23db 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -1,5 +1,5 @@ /* - * 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 @@ -14,15 +14,20 @@ package org.springframework.amqp.rabbit.core; import java.io.IOException; +import java.util.Map; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIllegalStateException; import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils; import org.springframework.amqp.rabbit.connection.RabbitAccessor; @@ -33,6 +38,7 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.AMQP.Queue.DeclareOk; @@ -73,9 +79,10 @@ import com.rabbitmq.client.GetResponse; * @author Mark Pollack * @author Mark Fisher * @author Dave Syer + * @author Gary Russell * @since 1.0 */ -public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { +public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, MessageListener { private static final String DEFAULT_EXCHANGE = ""; // alias for amq.direct default exchange @@ -98,7 +105,15 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter(); - private String encoding = DEFAULT_ENCODING; + private volatile String encoding = DEFAULT_ENCODING; + + private volatile Queue replyQueue; + + private final Map> replyHolder = new ConcurrentHashMap>(); + + public static final String STACKED_CORRELATION_HEADER = "spring_reply_correlation"; + + public static final String STACKED_REPLY_TO_HEADER = "spring_reply_to"; /** * Convenient constructor for use with setter injection. Don't forget to set the connection factory. @@ -164,6 +179,17 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { this.encoding = encoding; } + + /** + * A queue for replies; if not provided, a temporary exclusive, auto-delete queue will + * be used for each reply. + * + * @param replyQueue the replyQueue to set + */ + public void setReplyQueue(Queue replyQueue) { + this.replyQueue = replyQueue; + } + /** * Specify the timeout in milliseconds to be used when waiting for a reply Message when using one of the * sendAndReceive methods. The default value is defined as {@link #DEFAULT_REPLY_TIMEOUT}. A negative value @@ -358,6 +384,15 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { * @return the message that is received in reply */ protected Message doSendAndReceive(final String exchange, final String routingKey, final Message message) { + if (this.replyQueue == null) { + return doSendAndReceiveWithTemporary(exchange, routingKey, message); + } + else { + return doSendAndReceiveWithFixed(exchange, routingKey, message); + } + } + + protected Message doSendAndReceiveWithTemporary(final String exchange, final String routingKey, final Message message) { Message replyMessage = this.execute(new ChannelCallback() { public Message doInRabbit(Channel channel) throws Exception { final SynchronousQueue replyHandoff = new SynchronousQueue(); @@ -400,6 +435,50 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { return replyMessage; } + protected Message doSendAndReceiveWithFixed(final String exchange, final String routingKey, final Message message) { + Message replyMessage = this.execute(new ChannelCallback() { + public Message doInRabbit(Channel channel) throws Exception { + final LinkedBlockingQueue replyHandoff = new LinkedBlockingQueue(); + String messageTag = UUID.randomUUID().toString(); + RabbitTemplate.this.replyHolder.put(messageTag, replyHandoff); + + String replyTo = message.getMessageProperties().getReplyTo(); + if (StringUtils.hasLength(replyTo) && logger.isDebugEnabled()) { + logger.debug("Dropping replyTo header:" + replyTo + + " in favor of template's configured reply-queue:" + + RabbitTemplate.this.replyQueue.getName()); + } + String springReplyTo = (String) message.getMessageProperties() + .getHeaders().get(STACKED_REPLY_TO_HEADER); + message.getMessageProperties().setHeader( + STACKED_REPLY_TO_HEADER, + pushHeaderValue(replyTo, + springReplyTo)); + message.getMessageProperties().setReplyTo(RabbitTemplate.this.replyQueue.getName()); + String correlation = (String) message.getMessageProperties() + .getHeaders().get(STACKED_CORRELATION_HEADER); + if (StringUtils.hasLength(correlation)) { + message.getMessageProperties().setHeader( + STACKED_CORRELATION_HEADER, + pushHeaderValue(messageTag, correlation)); + } else { + message.getMessageProperties().setHeader( + "spring_reply_correlation", messageTag); + } + + if (logger.isDebugEnabled()) { + logger.debug("Sending message with tag " + messageTag); + } + doSend(channel, exchange, routingKey, message); + Message reply = (replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout, + TimeUnit.MILLISECONDS); + RabbitTemplate.this.replyHolder.remove(messageTag); + return reply; + } + }); + return replyMessage; + } + public T execute(ChannelCallback action) { Assert.notNull(action, "Callback object must not be null"); RabbitResourceHolder resourceHolder = getTransactionalResourceHolder(); @@ -483,4 +562,79 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { return name; } + public void onMessage(Message message) { + String messageTag = (String) message.getMessageProperties() + .getHeaders().get(STACKED_CORRELATION_HEADER); + if (messageTag == null) { + logger.error("No correlation header in reply"); + return; + } + PoppedHeader poppedHeaderValue = popHeaderValue(messageTag); + messageTag = poppedHeaderValue.getPoppedValue(); + message.getMessageProperties().setHeader(STACKED_CORRELATION_HEADER, + poppedHeaderValue.getNewValue()); + String springReplyTo = (String) message.getMessageProperties() + .getHeaders().get(STACKED_REPLY_TO_HEADER); + if (springReplyTo != null) { + poppedHeaderValue = popHeaderValue(springReplyTo); + springReplyTo = poppedHeaderValue.getNewValue(); + message.getMessageProperties().setHeader(STACKED_REPLY_TO_HEADER, springReplyTo); + message.getMessageProperties().setReplyTo(null); + } + LinkedBlockingQueue queue = this.replyHolder.get(messageTag); + if (queue == null) { + if (logger.isWarnEnabled()) { + logger.warn("Reply received after timeout for " + messageTag); + } + return; + } + queue.add(message); + if (logger.isDebugEnabled()) { + logger.debug("Reply received for " + messageTag); + } + } + + private String pushHeaderValue(String newValue, String oldValue) { + if (oldValue == null) { + return newValue; + } + else { + return newValue + ":" + oldValue; + } + } + + private PoppedHeader popHeaderValue(String value) { + int index = value.indexOf(":"); + if (index < 0) { + return new PoppedHeader(value, null); + } + else { + return new PoppedHeader(value.substring(0, index), value.substring(index+1)); + } + } + + private static class PoppedHeader { + + private final String poppedValue; + + private final String newValue; + + public PoppedHeader(String poppedValue, String newValue) { + this.poppedValue = poppedValue; + if (StringUtils.hasLength(newValue)) { + this.newValue = newValue; + } + else { + this.newValue = null; + } + } + + public String getPoppedValue() { + return poppedValue; + } + + public String getNewValue() { + return newValue; + } + } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 06c0d04b..47a3d4e2 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -72,6 +72,8 @@ public class BlockingQueueConsumer { private final AtomicBoolean cancelled = new AtomicBoolean(false); + private final AtomicBoolean cancelReceived = new AtomicBoolean(false); + private final AcknowledgeMode acknowledgeMode; private final ConnectionFactory connectionFactory; @@ -183,7 +185,11 @@ public class BlockingQueueConsumer { logger.debug("Retrieving delivery for " + this); } checkShutdown(); - return handle(queue.poll(timeout, TimeUnit.MILLISECONDS)); + Message message = handle(queue.poll(timeout, TimeUnit.MILLISECONDS)); + if (message == null && cancelReceived.get()) { + throw new ConsumerCancelledException(); + } + return message; } public void start() throws AmqpException { @@ -195,20 +201,36 @@ public class BlockingQueueConsumer { this.consumer = new InternalConsumer(channel); this.deliveryTags.clear(); this.activeObjectCounter.add(this); - try { - if (!acknowledgeMode.isAutoAck()) { - // Set basicQos before calling basicConsume (otherwise if we are not acking the broker - // will send blocks of 100 messages) - channel.basicQos(prefetchCount); + int passiveDeclareTries = 3; // mirrored queue might be being moved + do { + try { + if (!acknowledgeMode.isAutoAck()) { + // Set basicQos before calling basicConsume (otherwise if we are not acking the broker + // will send blocks of 100 messages) + channel.basicQos(prefetchCount); + } + for (int i = 0; i < queues.length; i++) { + channel.queueDeclarePassive(queues[i]); + } + passiveDeclareTries = 0; + } catch (IOException e) { + if (passiveDeclareTries > 0) { + if (logger.isWarnEnabled()) { + logger.warn("Reconnect failed; retries left=" + (passiveDeclareTries-1), e); + try { + Thread.sleep(5000); + } catch (InterruptedException e1) { + Thread.currentThread().interrupt(); + } + } + } else { + this.activeObjectCounter.release(this); + throw new FatalListenerStartupException("Cannot prepare queue for listener. " + + "Either the queue doesn't exist or the broker will not allow us to use it.", e); + } } - for (int i = 0; i < queues.length; i++) { - channel.queueDeclarePassive(queues[i]); - } - } catch (IOException e) { - this.activeObjectCounter.release(this); - throw new FatalListenerStartupException("Cannot prepare queue for listener. " - + "Either the queue doesn't exist or the broker will not allow us to use it.", e); - } + } while (passiveDeclareTries-- > 0); + try { for (int i = 0; i < queues.length; i++) { channel.basicConsume(queues[i], acknowledgeMode.isAutoAck(), consumer); @@ -223,13 +245,23 @@ public class BlockingQueueConsumer { public void stop() { cancelled.set(true); - if (consumer != null && consumer.getChannel() != null && consumer.getConsumerTag() != null) { - RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), transactional); + if (consumer != null && consumer.getChannel() != null && consumer.getConsumerTag() != null + && !this.cancelReceived.get()) { + try { + RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), transactional); + } catch (Exception e) { + if (logger.isDebugEnabled()) { + logger.debug("Error closing consumer", e); + } + } + } + if (logger.isDebugEnabled()) { + logger.debug("Closing Rabbit Channel: " + channel); } - logger.debug("Closing Rabbit Channel: " + channel); // This one never throws exceptions... RabbitUtils.closeChannel(channel); deliveryTags.clear(); + consumer = null; } private class InternalConsumer extends DefaultConsumer { @@ -248,6 +280,14 @@ public class BlockingQueueConsumer { deliveryTags.clear(); } + @Override + public void handleCancel(String consumerTag) throws IOException { + if (logger.isWarnEnabled()) { + logger.warn("Cancel received"); + } + cancelReceived.set(true); + } + @Override public void handleCancelOk(String consumerTag) { if (logger.isDebugEnabled()) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConsumerCancelledException.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConsumerCancelledException.java new file mode 100644 index 00000000..4936c43f --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConsumerCancelledException.java @@ -0,0 +1,30 @@ +/* + * 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. + */ +package org.springframework.amqp.rabbit.listener; + +/** + * Thrown when the broker cancels the consumer and the message + * queue is drained. + * + * @author Gary Russell + * @since 1.0.1 + * + */ +public class ConsumerCancelledException extends RuntimeException { + + private static final long serialVersionUID = 1L; + +} diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd index 03d44624..664e1116 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd @@ -56,6 +56,18 @@ ]]> + + + element. + ]]> + + + + + + + @@ -338,6 +350,7 @@ + - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - - - - - - - - - - - - - - - + + + + + - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + - - - - - - - - - - + + + + + + + + + + - - - - - - - - - - - - + + + + + + + + + + + + - - - - - - - - - - + + + + + + + + + + - - - - - + + + + + - - - - - + + + + + - - - - - + + + + + - - - - - + + + + + - - - - - + + + + + - - - - - + + + + + - - - - + ]]> + + + @@ -629,6 +649,17 @@ ]]> + + + + used to receive asynchronous replies on the reply-channel the + child element is disallowed because the template itself is the listener. + ]]> + + + + + + for replies; optional; if not supplied, methods expecting replies + will use a temporary, exclusive, auto-delete queue. + ]]> + + + + + + + @@ -733,6 +777,13 @@ ]]> + + + + + args = (Map) ctx.getBean("args"); + assertEquals("bar", args.get("foo")); + + assertEquals("qux", queue1.getArguments().get("baz")); + assertEquals("bar", queue2.getArguments().get("foo")); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserTests.java index 4116b3e4..a54433f2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserTests.java @@ -100,6 +100,13 @@ public class QueueParserTests { assertEquals("spam", queue.getArguments().get("foo")); } + @Test + public void testReferencedArgumentsQueue() throws Exception { + Queue queue = beanFactory.getBean("referencedArguments", Queue.class); + assertNotNull(queue); + assertEquals("qux", queue.getArguments().get("baz")); + } + @Test(expected=BeanDefinitionStoreException.class) public void testIllegalAnonymousQueue() throws Exception { beanFactory = new XmlBeanFactory(new ClassPathResource(getClass().getSimpleName() diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java index d8de26a7..714d0558 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java @@ -14,19 +14,24 @@ package org.springframework.amqp.rabbit.config; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import org.junit.Before; import org.junit.Test; import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.converter.SerializerMessageConverter; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.xml.XmlBeanFactory; import org.springframework.core.io.ClassPathResource; /** * * @author Dave Syer + * @author Gary Russell * */ public final class TemplateParserTests { @@ -50,5 +55,20 @@ public final class TemplateParserTests { assertNotNull(template); assertTrue(template.getMessageConverter() instanceof SerializerMessageConverter); } - + + @Test + public void testWithReplyQ() throws Exception { + RabbitTemplate template = beanFactory.getBean("withReplyQ", RabbitTemplate.class); + assertNotNull(template); + DirectFieldAccessor dfa = new DirectFieldAccessor(template); + Queue queue = (Queue) dfa.getPropertyValue("replyQueue"); + assertNotNull(queue); + Queue queueBean = beanFactory.getBean("reply.queue", Queue.class); + assertSame(queueBean, queue); + SimpleMessageListenerContainer container = beanFactory.getBean("withReplyQ.replyListener", SimpleMessageListenerContainer.class); + assertNotNull(container); + dfa = new DirectFieldAccessor(container); + assertSame(template, dfa.getPropertyValue("messageListener")); + } + } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java new file mode 100644 index 00000000..f304c7be --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java @@ -0,0 +1,240 @@ +/* + * 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. + */ +package org.springframework.amqp.rabbit.core; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; +import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.util.ReflectionUtils; + +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +/** + * @author Gary Russell + * @since 1.0.1 + * + */ +public class RabbitTemplateHeaderTests { + + @Test + public void testPushPop() throws Exception { + RabbitTemplate template = new RabbitTemplate(); + Method pushHeader = RabbitTemplate.class.getDeclaredMethod("pushHeaderValue", String.class, String.class); + ReflectionUtils.makeAccessible(pushHeader); + String header = (String) ReflectionUtils.invokeMethod(pushHeader, template, "a", null); + assertEquals("a", header); + header = (String) ReflectionUtils.invokeMethod(pushHeader, template, "b", header); + assertEquals("b:a", header); + header = (String) ReflectionUtils.invokeMethod(pushHeader, template, "c", header); + assertEquals("c:b:a", header); + + Method popHeader = RabbitTemplate.class.getDeclaredMethod("popHeaderValue", String.class); + ReflectionUtils.makeAccessible(popHeader); + Object poppedHeader = ReflectionUtils.invokeMethod(popHeader, template, header); + DirectFieldAccessor dfa = new DirectFieldAccessor(poppedHeader); + assertEquals("c", dfa.getPropertyValue("poppedValue")); + poppedHeader = ReflectionUtils.invokeMethod(popHeader, template, dfa.getPropertyValue("newValue")); + dfa = new DirectFieldAccessor(poppedHeader); + assertEquals("b", dfa.getPropertyValue("poppedValue")); + poppedHeader = ReflectionUtils.invokeMethod(popHeader, template, dfa.getPropertyValue("newValue")); + dfa = new DirectFieldAccessor(poppedHeader); + assertEquals("a", dfa.getPropertyValue("poppedValue")); + assertNull(dfa.getPropertyValue("newValue")); + } + + @Test + public void testReplyToOneDeep() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + Queue replyQueue = new Queue("new.replyTo"); + template.setReplyQueue(replyQueue); + + MessageProperties messageProperties = new MessageProperties(); + messageProperties.setReplyTo("replyTo1"); + Message message = new Message("Hello, world!".getBytes(), messageProperties); + final List props = new ArrayList(); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) throws Throwable { + BasicProperties basicProps = (BasicProperties) invocation.getArguments()[4]; + props.add(basicProps); + MessageProperties springProps = new DefaultMessagePropertiesConverter() + .toMessageProperties(basicProps, null, "UTF-8"); + Message replyMessage = new Message("!dlrow olleH".getBytes(), springProps); + template.onMessage(replyMessage); + return null; + }} + ).when(mockChannel).basicPublish(Mockito.any(String.class), + Mockito.any(String.class), Mockito.anyBoolean(), + Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); + Message reply = template.sendAndReceive(message); + + assertEquals(1, props.size()); + BasicProperties basicProperties = props.get(0); + assertEquals("new.replyTo", basicProperties.getReplyTo()); + assertEquals("replyTo1", basicProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER)); + assertNotNull(basicProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER)); + + } + + @Test + public void testReplyToTwoDeep() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + Queue replyQueue = new Queue("new.replyTo"); + template.setReplyQueue(replyQueue); + + MessageProperties messageProperties = new MessageProperties(); + messageProperties.setReplyTo("replyTo2"); + messageProperties.setHeader(RabbitTemplate.STACKED_REPLY_TO_HEADER, "replyTo1"); + messageProperties.setHeader(RabbitTemplate.STACKED_CORRELATION_HEADER, "a"); + Message message = new Message("Hello, world!".getBytes(), messageProperties); + final List props = new ArrayList(); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) throws Throwable { + BasicProperties basicProps = (BasicProperties) invocation.getArguments()[4]; + props.add(basicProps); + MessageProperties springProps = new DefaultMessagePropertiesConverter() + .toMessageProperties(basicProps, null, "UTF-8"); + Message replyMessage = new Message("!dlrow olleH".getBytes(), springProps); + template.onMessage(replyMessage); + return null; + }} + ).when(mockChannel).basicPublish(Mockito.any(String.class), + Mockito.any(String.class), Mockito.anyBoolean(), + Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); + Message reply = template.sendAndReceive(message); + + assertEquals(1, props.size()); + BasicProperties basicProperties = props.get(0); + assertEquals("new.replyTo", basicProperties.getReplyTo()); + assertEquals("replyTo2:replyTo1", basicProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER)); + assertTrue(((String)basicProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER)).endsWith(":a")); + + assertEquals("replyTo1", reply.getMessageProperties().getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER)); + assertEquals("a", reply.getMessageProperties().getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER)); + } + + @Test + public void testReplyToThreeDeep() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + Queue replyQueue = new Queue("new.replyTo"); + template.setReplyQueue(replyQueue); + + MessageProperties messageProperties = new MessageProperties(); + messageProperties.setReplyTo("replyTo2"); + messageProperties.setHeader(RabbitTemplate.STACKED_REPLY_TO_HEADER, "replyTo1"); + messageProperties.setHeader(RabbitTemplate.STACKED_CORRELATION_HEADER, "a"); + Message message = new Message("Hello, world!".getBytes(), messageProperties); + final List props = new ArrayList(); + final AtomicInteger count = new AtomicInteger(); + final List nestedReplyTo = new ArrayList(); + final List nestedReplyStack = new ArrayList(); + final List nestedCorrelation = new ArrayList(); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) throws Throwable { + BasicProperties basicProps = (BasicProperties) invocation.getArguments()[4]; + props.add(basicProps); + MessageProperties springProps = new DefaultMessagePropertiesConverter() + .toMessageProperties(basicProps, null, "UTF-8"); + Message replyMessage = new Message("!dlrow olleH".getBytes(), springProps); + if (count.incrementAndGet() < 2) { + Message anotherMessage = new Message("Second".getBytes(), springProps); + replyMessage = template.sendAndReceive(anotherMessage); + nestedReplyTo.add(replyMessage.getMessageProperties().getReplyTo()); + nestedReplyStack.add((String) replyMessage + .getMessageProperties().getHeaders() + .get(RabbitTemplate.STACKED_REPLY_TO_HEADER)); + nestedCorrelation.add((String) replyMessage + .getMessageProperties().getHeaders() + .get(RabbitTemplate.STACKED_CORRELATION_HEADER)); + } + template.onMessage(replyMessage); + return null; + }} + ).when(mockChannel).basicPublish(Mockito.any(String.class), + Mockito.any(String.class), Mockito.anyBoolean(), + Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); + Message reply = template.sendAndReceive(message); + assertNotNull(reply); + + assertEquals(2, props.size()); + BasicProperties basicProperties = props.get(0); + assertEquals("new.replyTo", basicProperties.getReplyTo()); + assertEquals("replyTo2:replyTo1", basicProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER)); + assertTrue(((String)basicProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER)).endsWith(":a")); + + basicProperties = props.get(1); + assertEquals("new.replyTo", basicProperties.getReplyTo()); + assertEquals("new.replyTo:replyTo2:replyTo1", basicProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER)); + assertTrue(((String)basicProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER)).endsWith(":a")); + + assertEquals("replyTo1", reply.getMessageProperties().getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER)); + assertEquals("a", reply.getMessageProperties().getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER)); + + assertEquals(1, nestedReplyTo.size()); + assertEquals(1, nestedReplyStack.size()); + assertEquals(1, nestedCorrelation.size()); + assertEquals("replyTo2:replyTo1", nestedReplyStack.get(0)); + assertTrue(nestedCorrelation.get(0).endsWith(":a")); + + } +} diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml index 0d9dd2c7..9a736300 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml @@ -29,4 +29,7 @@ + + diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests-context.xml new file mode 100644 index 00000000..195f6305 --- /dev/null +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests-context.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + + diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserPlaceholderTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserPlaceholderTests-context.xml index 481980cf..f42edfbe 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserPlaceholderTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserPlaceholderTests-context.xml @@ -38,4 +38,10 @@ + + + + + + diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserTests-context.xml index 3a0565ee..4e04e89a 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserTests-context.xml @@ -28,4 +28,10 @@ + + + + + + diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml index 275fa881..520f18ba 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml @@ -1,7 +1,7 @@ @@ -10,7 +10,17 @@ encoding="UTF-8" exchange="foo" queue="bar" routing-key="spam" message-converter="converter" reply-timeout="1000" /> - + + + + + + + + + + + diff --git a/src/docbkx/amqp.xml b/src/docbkx/amqp.xml index fd894111..4f09cc0e 100644 --- a/src/docbkx/amqp.xml +++ b/src/docbkx/amqp.xml @@ -296,6 +296,17 @@ Connection connection = connectionFactory.createConnection();]]>]]> + + Host and port attributes can be provided using the namespace + ]]> + + + Alternatively, if running in a clustered environment, use the addresses + attribute. + ]]> +
@@ -685,6 +696,23 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]>Similar request/reply methods are also available where the MessageConverter is applied to both the request and reply. Those methods are named convertSendAndReceive. See the Javadoc of AmqpTemplate for more detail. + + By default, a new temporary queue is used for each reply. However, a single reply queue can be configured on the template, + which allows you to set arguments on that queue (such as 'ha_args="all"' for mirrored queues). In this case, however, + you must also provide a <reply-listener/> sub element. This element provides a listener container for the + reply queue, with the template being the listener. All of the attributes + allowed on a <listener-container/> are allowed on the element, except for connection-factory and + message-converter, which are inherited from the template's configuration. + + + +]]> + + + While the container and template share a connection factory, they do not share a channel and therefore requests + and replies are not performed within the same transaction (if transactional). +
@@ -997,14 +1025,18 @@ public class ExampleExternalTransactionAmqpConfiguration {
-
- Message Listener Container Features +
+ Message Listener Container Configuration There are quite a few options for configuring a SimpleMessageListenerContainer related to transactions and quality of service, and some of them interact with each other. + When configuring with the XML namespace, the convention is to + use hyphenated attributes rather than camel case; for example, for + property 'connectionFactory', the XML equivalent is 'connection-factory'. + Configuration options for a message listener container @@ -1105,6 +1137,17 @@ public class ExampleExternalTransactionAmqpConfiguration { broker is ready. + + phase + + When autoStartup is true, the lifecycle phase within + which this container should start and stop. The lower the + value the earlier this container will start and the later it + will stop. The default is Integer.MAX_VALUE meaning the + container will start as late as possible and stop as + soon as possible. + + adviceChain @@ -1115,6 +1158,46 @@ public class ExampleExternalTransactionAmqpConfiguration { the CachingConnectionFactory, as long as the broker is still alive. + + + taskExecutor + + A reference to a Spring TaskExecutor (or standard JDK 1.5+ + Executor) for executing listener invokers. Default is a + SimpleAsyncTaskExecutor, using internally managed threads. + + + + errorHandler + + A reference to an ErrorHandler strategy for handling any + uncaught Exceptions that may occur during the execution of the + MessageListener. + + + + concurrency + + The number of concurrent consumers to start for each + listener. + + + + connectionFactory + + A reference to the connectionFactory; when configuring + using the XML namespace, the default referenced bean name + is "rabbitConnectionFactory". + + + + messageConverter + + A reference to the MessageConverter strategy for + converting AMQP Messages to listener method arguments + for any referenced 'listener' that is a POJO. Default is + a SimpleMessageConverter. +
diff --git a/src/docbkx/index.xml b/src/docbkx/index.xml index aef2f40f..aeed365e 100644 --- a/src/docbkx/index.xml +++ b/src/docbkx/index.xml @@ -23,10 +23,14 @@ Dave Syer + + Gary + Russell + - Copyright © 2010-2011 + Copyright © 2010-2012 Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and