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..e9fbc671 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"; @@ -41,6 +43,10 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { private static final String EXECUTOR_ATTRIBUTE = "executor"; + private static final String PUBLISHER_CONFIRMS = "publisher-confirms"; + + private static final String PUBLISHER_RETURNS = "publisher-returns"; + @Override protected Class getBeanClass(Element element) { return CachingConnectionFactory.class; @@ -58,7 +64,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 +77,10 @@ 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); + NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_CONFIRMS); + NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_RETURNS); + } } 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..5e69171c 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 @@ -13,15 +13,23 @@ package org.springframework.amqp.rabbit.config; +import java.util.List; + 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.springframework.util.xml.DomUtils; 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 +49,18 @@ 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"; + + private static final String MANDATORY_ATTRIBUTE = "mandatory"; + + private static final String IMMEDIATE_ATTRIBUTE = "immediate"; + + private static final String RETURN_CALLBACK_ATTRIBUTE = "return-callback"; + + private static final String CONFIRM_CALLBACK_ATTRIBUTE = "confirm-callback"; + @Override protected Class getBeanClass(Element element) { return RabbitTemplate.class; @@ -53,7 +73,7 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { @Override protected boolean shouldGenerateIdAsFallback() { - return true; + return false; } @Override @@ -77,7 +97,54 @@ 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); + NamespaceUtils.setValueIfAttributeDefined(builder, element, MANDATORY_ATTRIBUTE); + NamespaceUtils.setValueIfAttributeDefined(builder, element, IMMEDIATE_ATTRIBUTE); + NamespaceUtils.setReferenceIfAttributeDefined(builder, element, RETURN_CALLBACK_ATTRIBUTE); + NamespaceUtils.setReferenceIfAttributeDefined(builder, element, CONFIRM_CALLBACK_ATTRIBUTE); + BeanDefinition replyContainer = null; + Element childElement = null; + List childElements = DomUtils.getChildElementsByTagName(element, LISTENER_ELEMENT); + if (childElements.size() > 0) { + childElement = childElements.get(0); + } + 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 BeanDefinition parseListener(Element childElement, Element element, + ParserContext parserContext) { + 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/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 63f51861..062c9493 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -13,6 +13,7 @@ package org.springframework.amqp.rabbit.connection; +import java.io.IOException; import java.lang.reflect.InvocationHandler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; @@ -21,6 +22,8 @@ import java.util.LinkedList; import java.util.List; import org.springframework.amqp.AmqpException; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannel; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -57,6 +60,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { private ChannelCachingConnectionProxy connection; + private volatile boolean publisherConfirms; + /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); @@ -118,6 +123,14 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { return this.channelCacheSize; } + public boolean isPublisherConfirms() { + return publisherConfirms; + } + + public void setPublisherConfirms(boolean publisherConfirms) { + this.publisherConfirms = publisherConfirms; + } + public void setConnectionListeners(List listeners) { super.setConnectionListeners(listeners); // If the connection is already alive we assume that the new listeners want to be notified @@ -159,8 +172,15 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { logger.debug("Creating cached Rabbit Channel from " + targetChannel); } getChannelListener().onCreate(targetChannel, transactional); + Class[] interfaces; + if (this.publisherConfirms) { + interfaces = new Class[] { ChannelProxy.class, PublisherCallbackChannel.class }; + } + else { + interfaces = new Class[] { ChannelProxy.class }; + } return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class.getClassLoader(), - new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(targetChannel, channelList, + interfaces, new CachedChannelInvocationHandler(targetChannel, channelList, transactional)); } @@ -170,7 +190,20 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { // Use createConnection here not doCreateConnection so that the old one is properly disposed createConnection(); } - return this.connection.createBareChannel(transactional); + Channel channel = this.connection.createBareChannel(transactional); + if (this.publisherConfirms) { + try { + channel.confirmSelect(); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + if (!(channel instanceof PublisherCallbackChannelImpl)) { + channel = new PublisherCallbackChannelImpl(channel); + } + } + // TODO returns + return channel; } public final Connection createConnection() throws AmqpException { 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..aecd106b 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,27 +14,43 @@ package org.springframework.amqp.rabbit.core; import java.io.IOException; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.SortedMap; 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; import org.springframework.amqp.rabbit.connection.RabbitResourceHolder; import org.springframework.amqp.rabbit.connection.RabbitUtils; +import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; +import org.springframework.amqp.rabbit.support.PendingConfirm; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannel; 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.BasicProperties; import com.rabbitmq.client.AMQP.Queue.DeclareOk; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; @@ -73,9 +89,11 @@ 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, + PublisherCallbackChannel.Listener { private static final String DEFAULT_EXCHANGE = ""; // alias for amq.direct default exchange @@ -98,7 +116,27 @@ 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>(); + + private volatile ConfirmCallback confirmCallback; + + private volatile ReturnCallback returnCallback; + + private final Map> pendingConfirms = new ConcurrentHashMap>(); + + private volatile boolean mandatory; + + private volatile boolean immediate; + + private final String uuid = UUID.randomUUID().toString(); + + 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 +202,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 @@ -213,6 +262,55 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { return this.messageConverter; } + public void setConfirmCallback(ConfirmCallback confirmCallback) { + Assert.state(this.confirmCallback == null || this.confirmCallback == confirmCallback, + "Only one ConfirmCallback is supported by each RabbitTemplate"); + this.confirmCallback = confirmCallback; + } + + public void setReturnCallback(ReturnCallback returnCallback) { + Assert.state(this.returnCallback == null || this.returnCallback == returnCallback, + "Only one ReturnCallback is supported by each RabbitTemplate"); + this.returnCallback = returnCallback; + } + + public void setMandatory(boolean mandatory) { + this.mandatory = mandatory; + } + + public void setImmediate(boolean immediate) { + this.immediate = immediate; + } + + /** + * Gets unconfirmed correlatiom data older than age and removes them. + * @param age in millseconds + * @return the collection of correlation data for which confirms have + * not been received. + */ + public Collection getUnconfirmed(long age) { + Set unconfirmed = new HashSet(); + synchronized (this.pendingConfirms) { + long threshold = System.currentTimeMillis() - age; + for (Entry> channelPendingConfirmEntry : this.pendingConfirms.entrySet()) { + SortedMap channelPendingConfirms = channelPendingConfirmEntry.getValue(); + Iterator> iterator = channelPendingConfirms.entrySet().iterator(); + PendingConfirm pendingConfirm; + while (iterator.hasNext()) { + pendingConfirm = iterator.next().getValue(); + if (pendingConfirm.getTimestamp() < threshold) { + unconfirmed.add(pendingConfirm.getCorrelationData()); + iterator.remove(); + } + else { + break; + } + } + } + } + return unconfirmed.size() > 0 ? unconfirmed : null; + } + public void send(Message message) throws AmqpException { send(this.exchange, this.routingKey, message); } @@ -222,24 +320,42 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { } public void send(final String exchange, final String routingKey, final Message message) throws AmqpException { + this.send(exchange, routingKey, message, null); + } + + public void send(final String exchange, final String routingKey, + final Message message, final CorrelationData correlationData) + throws AmqpException { execute(new ChannelCallback() { public Object doInRabbit(Channel channel) throws Exception { - doSend(channel, exchange, routingKey, message); + doSend(channel, exchange, routingKey, message, correlationData); return null; } }); } public void convertAndSend(Object object) throws AmqpException { - convertAndSend(this.exchange, this.routingKey, object); + convertAndSend(this.exchange, this.routingKey, object, (CorrelationData) null); + } + + public void correlationconvertAndSend(Object object, CorrelationData correlationData) throws AmqpException { + convertAndSend(this.exchange, this.routingKey, object, correlationData); } public void convertAndSend(String routingKey, final Object object) throws AmqpException { - convertAndSend(this.exchange, routingKey, object); + convertAndSend(this.exchange, routingKey, object, (CorrelationData) null); + } + + public void convertAndSend(String routingKey, final Object object, CorrelationData correlationData) throws AmqpException { + convertAndSend(this.exchange, routingKey, object, correlationData); } public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException { - send(exchange, routingKey, getRequiredMessageConverter().toMessage(object, new MessageProperties())); + convertAndSend(exchange, routingKey, object, (CorrelationData) null); + } + + public void convertAndSend(String exchange, String routingKey, final Object object, CorrelationData corrationData) throws AmqpException { + send(exchange, routingKey, getRequiredMessageConverter().toMessage(object, new MessageProperties()), corrationData); } public void convertAndSend(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException { @@ -248,14 +364,25 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { public void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor) throws AmqpException { - convertAndSend(this.exchange, routingKey, message, messagePostProcessor); + convertAndSend(this.exchange, routingKey, message, messagePostProcessor, null); + } + + public void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor, + CorrelationData correlationData) + throws AmqpException { + convertAndSend(this.exchange, routingKey, message, messagePostProcessor, correlationData); } public void convertAndSend(String exchange, String routingKey, final Object message, final MessagePostProcessor messagePostProcessor) throws AmqpException { + convertAndSend(exchange, routingKey, message, messagePostProcessor, null); + } + + public void convertAndSend(String exchange, String routingKey, final Object message, + final MessagePostProcessor messagePostProcessor, CorrelationData correlationData) throws AmqpException { Message messageToSend = getRequiredMessageConverter().toMessage(message, new MessageProperties()); messageToSend = messagePostProcessor.postProcessMessage(messageToSend); - send(exchange, routingKey, messageToSend); + send(exchange, routingKey, messageToSend, correlationData); } public Message receive() throws AmqpException { @@ -358,6 +485,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(); @@ -390,7 +526,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { } }; channel.basicConsume(replyTo, noAck, consumerTag, noLocal, exclusive, null, consumer); - doSend(channel, exchange, routingKey, message); + doSend(channel, exchange, routingKey, message, null); Message reply = (replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout, TimeUnit.MILLISECONDS); channel.basicCancel(consumerTag); @@ -400,10 +536,57 @@ 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, null); + 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(); Channel channel = resourceHolder.getChannel(); + if (this.confirmCallback != null || this.returnCallback != null) { + addListener(channel); + } try { if (logger.isDebugEnabled()) { logger.debug("Executing callback on RabbitMQ Channel: " + channel); @@ -428,7 +611,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { * @param message the Message to send * @throws IOException if thrown by RabbitMQ API methods */ - protected void doSend(Channel channel, String exchange, String routingKey, Message message) throws Exception { + protected void doSend(Channel channel, String exchange, String routingKey, Message message, + CorrelationData correlationData) throws Exception { if (logger.isDebugEnabled()) { logger.debug("Publishing message on exchange [" + exchange + "], routingKey = [" + routingKey + "]"); } @@ -442,10 +626,21 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { // try to send to configured routing key routingKey = this.routingKey; } - - channel.basicPublish(exchange, routingKey, false, false, - this.messagePropertiesConverter.fromMessageProperties(message.getMessageProperties(), encoding), - message.getBody()); + if (this.confirmCallback != null && channel instanceof PublisherCallbackChannel) { + PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel; + publisherCallbackChannel.addPendingConfirm(this, channel.getNextPublishSeqNo(), + new PendingConfirm(correlationData, System.currentTimeMillis())); + } + boolean mandatory = this.returnCallback == null ? false : this.mandatory; + boolean immediate = this.returnCallback == null ? false : this.immediate; + MessageProperties messageProperties = message.getMessageProperties(); + if (mandatory || immediate) { + messageProperties.getHeaders().put(PublisherCallbackChannel.RETURN_CORRELATION, this.uuid); + } + BasicProperties convertedMessageProperties = this.messagePropertiesConverter + .fromMessageProperties(messageProperties, encoding); + channel.basicPublish(exchange, routingKey, mandatory, immediate, + convertedMessageProperties, message.getBody()); // Check if commit needed if (isChannelLocallyTransacted(channel)) { // Transacted channel created by this template -> commit. @@ -483,4 +678,160 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { return name; } + private void addListener(Channel channel) { + if (channel instanceof PublisherCallbackChannel) { + PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel; + SortedMap pendingConfirms = publisherCallbackChannel.addListener(this); + if (!this.pendingConfirms.containsKey(channel)) { + this.pendingConfirms.put(channel, pendingConfirms); + if (logger.isDebugEnabled()) { + logger.debug("Added pending confirms for " + channel + " to map, size now " + this.pendingConfirms.size()); + } + } + } + else { + throw new IllegalStateException("When using publisher confirms, channels must be wrapped in a PublisherCallbackChannelImpl"); + } + } + + public void handleConfirm(PendingConfirm pendingConfirm, boolean ack) { + if (this.confirmCallback != null) { + this.confirmCallback.confirm(pendingConfirm.getCorrelationData(), ack); + } + else { + if (logger.isDebugEnabled()) { + logger.warn("Confirm received but no callback available"); + } + } + } + + public void handleReturn(int replyCode, + String replyText, + String exchange, + String routingKey, + BasicProperties properties, + byte[] body) + throws IOException + { + if (this.returnCallback == null) { + if (logger.isWarnEnabled()) { + logger.warn("Returned message but no callback available"); + } + } + else { + properties.getHeaders().remove(PublisherCallbackChannel.RETURN_CORRELATION); + MessageProperties messageProperties = messagePropertiesConverter.toMessageProperties( + properties, null, this.encoding); + Message returnedMessage = new Message(body, messageProperties); + this.returnCallback.returnedMessage(returnedMessage, + replyCode, replyText, exchange, routingKey); + } + } + + public boolean isConfirmListener() { + return this.confirmCallback != null; + } + + public boolean isReturnListener() { + return this.returnCallback != null; + } + + public void removePendingConfirmsReference(Channel channel, + SortedMap unconfirmed) { + this.pendingConfirms.remove(channel); + if (logger.isDebugEnabled()) { + logger.debug("Removed pending confirms for " + channel + " from map, size now " + this.pendingConfirms.size()); + } + } + + public String getUUID() { + return this.uuid; + } + + 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; + } + } + + public static interface ConfirmCallback { + + void confirm(CorrelationData correlationData, boolean ack); + } + + public static interface ReturnCallback { + + void returnedMessage(Message message, int replyCode, String replyText, + String exchange, String routingKey); + } } 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/java/org/springframework/amqp/rabbit/support/CorrelationData.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java new file mode 100644 index 00000000..f88517ba --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java @@ -0,0 +1,45 @@ +/* + * 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.support; + +import org.springframework.amqp.rabbit.core.RabbitTemplate; + +/** + * Base class for correlating publisher confirms to sent messages. + * Use the {@link RabbitTemplate} methods that include one of + * these as a parameter; when the publisher confirm is received, + * the CorrelationData is returned with the ack/nack. + * @author Gary Russell + * @since 1.0.1 + * + */ +public class CorrelationData { + + private String id; + + public CorrelationData(String id) { + this.id = id; + } + + public String getId() { + return id; + } + + @Override + public String toString() { + return "CorrelationData [id=" + id + "]"; + } +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java new file mode 100644 index 00000000..4f87fa8f --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java @@ -0,0 +1,55 @@ +/* + * 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.support; + +/** + * Instances of this object track pending publisher confirms. + * The timestamp allows the pending confirmation to be + * expired. It also holds {@link CorrelationData} for + * the client to correlate a confirm with a sent message. + * @author Gary Russell + * @since 1.0.1 + * + */ +public class PendingConfirm { + + private final CorrelationData correlationData; + + private final long timestamp; + + /** + * @param correlationId + * @param timestamp + */ + public PendingConfirm(CorrelationData correlationData, long timestamp) { + this.correlationData = correlationData; + this.timestamp = timestamp; + } + + public CorrelationData getCorrelationData() { + return correlationData; + } + + public long getTimestamp() { + return timestamp; + } + + @Override + public String toString() { + return "PendingConfirm [correlationData=" + correlationData + "]"; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java new file mode 100644 index 00000000..ef47302a --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java @@ -0,0 +1,104 @@ +/* + * 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.support; + +import java.io.IOException; +import java.util.SortedMap; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Channel; + +/** + * Instances of this interface support a single listener being + * registered for publisher confirms with multiple channels, + * by adding context to the callbacks. + * @author Gary Russell + * @since 1.0.1 + * + */ +public interface PublisherCallbackChannel extends Channel { + + static String RETURN_CORRELATION = "spring_return_correlation"; + + /** + * Adds a {@link Listener} and returns a reference to + * the pending confirms map for that listener's pending + * confirms, allowing the Listener to + * assess unconfirmed sends at any point in time. + * The client must NOT modify the contents of + * this array, and must synchronize on it when + * iterating over its collections. + * @param listener The Listener. + * @return A reference to pending confirms for the listener + */ + SortedMap addListener(Listener listener); + + /** + * Gets a reference to the current listener, or null. + * @return the Listener. + */ + boolean removeListener(Listener listener); + + /** + * Adds a pending confirmation to this channel's map. + * @param seq The key to the map. + * @param pendingConfirm The PendingConfirm object. + */ + void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm); + + /** + * Listeners implementing this interface can participate + * in publisher confirms received from multiple channels, + * by invoking addListener on each channel. Standard + * AMQP channels do not support a listener being + * registered on multiple channels. + */ + public static interface Listener { + + /** + * Invoked by the channel when a confirm is received. + * @param pendingConfirm The pending confirmation, containing + * correlation data. + * @param ack true when 'ack', false when 'nack'. + */ + void handleConfirm(PendingConfirm pendingConfirm, boolean ack); + + void handleReturn(int replyCode, + String replyText, + String exchange, + String routingKey, + AMQP.BasicProperties properties, + byte[] body) throws IOException; + + /** + * When called, this listener must remove all references to the + * pending confirm map. + * @param unconfirmed The pending confirm map. + */ + void removePendingConfirmsReference(Channel channel, SortedMap unconfirmed); + + /** + * Returns the UUID used to identify this Listener for returns. + * @return A string representation of the UUID. + */ + String getUUID(); + + boolean isConfirmListener(); + + boolean isReturnListener(); + } + +} \ No newline at end of file diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java new file mode 100644 index 00000000..d3c9a2a5 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java @@ -0,0 +1,559 @@ +/* + * 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.support; + +import java.io.IOException; +import java.util.Collections; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeoutException; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.util.Assert; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AMQP.Basic.RecoverOk; +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.AMQP.Channel.FlowOk; +import com.rabbitmq.client.AMQP.Exchange.BindOk; +import com.rabbitmq.client.AMQP.Exchange.DeclareOk; +import com.rabbitmq.client.AMQP.Exchange.DeleteOk; +import com.rabbitmq.client.AMQP.Exchange.UnbindOk; +import com.rabbitmq.client.AMQP.Queue.PurgeOk; +import com.rabbitmq.client.AMQP.Tx.CommitOk; +import com.rabbitmq.client.AMQP.Tx.RollbackOk; +import com.rabbitmq.client.AMQP.Tx.SelectOk; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Command; +import com.rabbitmq.client.ConfirmListener; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.Consumer; +import com.rabbitmq.client.FlowListener; +import com.rabbitmq.client.GetResponse; +import com.rabbitmq.client.Method; +import com.rabbitmq.client.ReturnListener; +import com.rabbitmq.client.ShutdownListener; +import com.rabbitmq.client.ShutdownSignalException; + +/** + * Channel wrapper to allow a single listener able to handle + * confirms from multiple channels. + * + * @author Gary Russell + * @since 1.0.1 + * + */ +public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final Channel delegate; + + private final Map listeners = new ConcurrentHashMap(); + + private final Map> pendingConfirms + = new ConcurrentHashMap>(); + + private final Map listenerForSeq = new ConcurrentHashMap(); + + public PublisherCallbackChannelImpl(Channel delegate) { + this.delegate = delegate; + } + +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// BEGIN PURE DELEGATE METHODS +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + public void addShutdownListener(ShutdownListener listener) { + this.delegate.addShutdownListener(listener); + } + + public void removeShutdownListener(ShutdownListener listener) { + this.delegate.removeShutdownListener(listener); + } + + public ShutdownSignalException getCloseReason() { + return this.delegate.getCloseReason(); + } + + public void notifyListeners() { + this.delegate.notifyListeners(); + } + + public boolean isOpen() { + return this.delegate.isOpen(); + } + + public int getChannelNumber() { + return this.delegate.getChannelNumber(); + } + + public Connection getConnection() { + return this.delegate.getConnection(); + } + + public void close(int closeCode, String closeMessage) throws IOException { + this.delegate.close(closeCode, closeMessage); + } + + public FlowOk flow(boolean active) throws IOException { + return this.delegate.flow(active); + } + + public FlowOk getFlow() { + return this.delegate.getFlow(); + } + + public void abort() throws IOException { + this.delegate.abort(); + } + + public void abort(int closeCode, String closeMessage) throws IOException { + this.delegate.abort(closeCode, closeMessage); + } + + public void addFlowListener(FlowListener listener) { + this.delegate.addFlowListener(listener); + } + + public boolean removeFlowListener(FlowListener listener) { + return this.delegate.removeFlowListener(listener); + } + + public void clearFlowListeners() { + this.delegate.clearFlowListeners(); + } + + public Consumer getDefaultConsumer() { + return this.delegate.getDefaultConsumer(); + } + + public void setDefaultConsumer(Consumer consumer) { + this.delegate.setDefaultConsumer(consumer); + } + + public void basicQos(int prefetchSize, int prefetchCount, boolean global) + throws IOException { + this.delegate.basicQos(prefetchSize, prefetchCount, global); + } + + public void basicQos(int prefetchCount) throws IOException { + this.delegate.basicQos(prefetchCount); + } + + public void basicPublish(String exchange, String routingKey, + BasicProperties props, byte[] body) throws IOException { + this.delegate.basicPublish(exchange, routingKey, props, body); + } + + public void basicPublish(String exchange, String routingKey, + boolean mandatory, boolean immediate, BasicProperties props, + byte[] body) throws IOException { + this.delegate.basicPublish(exchange, routingKey, mandatory, immediate, + props, body); + } + + public DeclareOk exchangeDeclare(String exchange, String type) + throws IOException { + return this.delegate.exchangeDeclare(exchange, type); + } + + public DeclareOk exchangeDeclare(String exchange, String type, + boolean durable) throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable); + } + + public DeclareOk exchangeDeclare(String exchange, String type, + boolean durable, boolean autoDelete, Map arguments) + throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable, autoDelete, + arguments); + } + + public DeclareOk exchangeDeclare(String exchange, String type, + boolean durable, boolean autoDelete, boolean internal, + Map arguments) throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable, autoDelete, + internal, arguments); + } + + public DeclareOk exchangeDeclarePassive(String name) throws IOException { + return this.delegate.exchangeDeclarePassive(name); + } + + public DeleteOk exchangeDelete(String exchange, boolean ifUnused) + throws IOException { + return this.delegate.exchangeDelete(exchange, ifUnused); + } + + public DeleteOk exchangeDelete(String exchange) throws IOException { + return this.delegate.exchangeDelete(exchange); + } + + public BindOk exchangeBind(String destination, String source, + String routingKey) throws IOException { + return this.delegate.exchangeBind(destination, source, routingKey); + } + + public BindOk exchangeBind(String destination, String source, + String routingKey, Map arguments) + throws IOException { + return this.delegate + .exchangeBind(destination, source, routingKey, arguments); + } + + public UnbindOk exchangeUnbind(String destination, String source, + String routingKey) throws IOException { + return this.delegate.exchangeUnbind(destination, source, routingKey); + } + + public UnbindOk exchangeUnbind(String destination, String source, + String routingKey, Map arguments) + throws IOException { + return this.delegate.exchangeUnbind(destination, source, routingKey, + arguments); + } + + public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclare() + throws IOException { + return this.delegate.queueDeclare(); + } + + public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclare(String queue, + boolean durable, boolean exclusive, boolean autoDelete, + Map arguments) throws IOException { + return this.delegate.queueDeclare(queue, durable, exclusive, autoDelete, + arguments); + } + + public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclarePassive( + String queue) throws IOException { + return this.delegate.queueDeclarePassive(queue); + } + + public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete(String queue) + throws IOException { + return this.delegate.queueDelete(queue); + } + + public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete(String queue, + boolean ifUnused, boolean ifEmpty) throws IOException { + return this.delegate.queueDelete(queue, ifUnused, ifEmpty); + } + + public com.rabbitmq.client.AMQP.Queue.BindOk queueBind(String queue, + String exchange, String routingKey) throws IOException { + return this.delegate.queueBind(queue, exchange, routingKey); + } + + public com.rabbitmq.client.AMQP.Queue.BindOk queueBind(String queue, + String exchange, String routingKey, Map arguments) + throws IOException { + return this.delegate.queueBind(queue, exchange, routingKey, arguments); + } + + public com.rabbitmq.client.AMQP.Queue.UnbindOk queueUnbind(String queue, + String exchange, String routingKey) throws IOException { + return this.delegate.queueUnbind(queue, exchange, routingKey); + } + + public com.rabbitmq.client.AMQP.Queue.UnbindOk queueUnbind(String queue, + String exchange, String routingKey, Map arguments) + throws IOException { + return this.delegate.queueUnbind(queue, exchange, routingKey, arguments); + } + + public PurgeOk queuePurge(String queue) throws IOException { + return this.delegate.queuePurge(queue); + } + + public GetResponse basicGet(String queue, boolean autoAck) + throws IOException { + return this.delegate.basicGet(queue, autoAck); + } + + public void basicAck(long deliveryTag, boolean multiple) throws IOException { + this.delegate.basicAck(deliveryTag, multiple); + } + + public void basicNack(long deliveryTag, boolean multiple, boolean requeue) + throws IOException { + this.delegate.basicNack(deliveryTag, multiple, requeue); + } + + public void basicReject(long deliveryTag, boolean requeue) + throws IOException { + this.delegate.basicReject(deliveryTag, requeue); + } + + public String basicConsume(String queue, Consumer callback) + throws IOException { + return this.delegate.basicConsume(queue, callback); + } + + public String basicConsume(String queue, boolean autoAck, Consumer callback) + throws IOException { + return this.delegate.basicConsume(queue, autoAck, callback); + } + + public String basicConsume(String queue, boolean autoAck, + String consumerTag, Consumer callback) throws IOException { + return this.delegate.basicConsume(queue, autoAck, consumerTag, callback); + } + + public String basicConsume(String queue, boolean autoAck, + String consumerTag, boolean noLocal, boolean exclusive, + Map arguments, Consumer callback) + throws IOException { + return this.delegate.basicConsume(queue, autoAck, consumerTag, noLocal, + exclusive, arguments, callback); + } + + public void basicCancel(String consumerTag) throws IOException { + this.delegate.basicCancel(consumerTag); + } + + public RecoverOk basicRecover() throws IOException { + return this.delegate.basicRecover(); + } + + public RecoverOk basicRecover(boolean requeue) throws IOException { + return this.delegate.basicRecover(requeue); + } + + @SuppressWarnings("deprecation") + public void basicRecoverAsync(boolean requeue) throws IOException { + this.delegate.basicRecoverAsync(requeue); + } + + public SelectOk txSelect() throws IOException { + return this.delegate.txSelect(); + } + + public CommitOk txCommit() throws IOException { + return this.delegate.txCommit(); + } + + public RollbackOk txRollback() throws IOException { + return this.delegate.txRollback(); + } + + public com.rabbitmq.client.AMQP.Confirm.SelectOk confirmSelect() + throws IOException { + return this.delegate.confirmSelect(); + } + + public long getNextPublishSeqNo() { + return this.delegate.getNextPublishSeqNo(); + } + + public boolean waitForConfirms() throws InterruptedException { + return this.delegate.waitForConfirms(); + } + + public boolean waitForConfirms(long timeout) throws InterruptedException, + TimeoutException { + return this.delegate.waitForConfirms(timeout); + } + + public void waitForConfirmsOrDie() throws IOException, InterruptedException { + this.delegate.waitForConfirmsOrDie(); + } + + public void waitForConfirmsOrDie(long timeout) throws IOException, + InterruptedException, TimeoutException { + this.delegate.waitForConfirmsOrDie(timeout); + } + + public void asyncRpc(Method method) throws IOException { + this.delegate.asyncRpc(method); + } + + public Command rpc(Method method) throws IOException { + return this.delegate.rpc(method); + } + + public void addConfirmListener(ConfirmListener listener) { + this.delegate.addConfirmListener(listener); + } + + public boolean removeConfirmListener(ConfirmListener listener) { + return this.delegate.removeConfirmListener(listener); + } + + public void clearConfirmListeners() { + this.delegate.clearConfirmListeners(); + } + + public void addReturnListener(ReturnListener listener) { + this.delegate.addReturnListener(listener); + } + + public boolean removeReturnListener(ReturnListener listener) { + return this.delegate.removeReturnListener(listener); + } + + public synchronized void clearReturnListeners() { + this.delegate.clearReturnListeners(); + } + +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// END PURE DELEGATE METHODS +////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + public void close() throws IOException { + this.delegate.close(); + for (Entry> entry : this.pendingConfirms.entrySet()) { + Listener listener = entry.getKey(); + listener.removePendingConfirmsReference(this, entry.getValue()); + } + this.pendingConfirms.clear(); + this.listenerForSeq.clear(); + } + + public synchronized SortedMap addListener(Listener listener) { + Assert.notNull(listener, "Listener cannot be null"); + if (this.listeners.size() == 0) { + this.delegate.addConfirmListener(this); + this.delegate.addReturnListener(this); + } + if (!this.listeners.values().contains(listener)){ + this.listeners.put(listener.getUUID(), listener); + this.pendingConfirms.put(listener, Collections.synchronizedSortedMap(new TreeMap())); + if (logger.isDebugEnabled()) { + logger.debug("Added listener " + listener); + } + } + return this.pendingConfirms.get(listener); + } + + public synchronized boolean removeListener(Listener listener) { + Listener mappedListener = this.listeners.remove(listener.getUUID()); + boolean result = mappedListener != null; + if (result && this.listeners.size() == 0) { + this.delegate.removeConfirmListener(this); + this.delegate.removeReturnListener(this); + } + Iterator> iterator = this.listenerForSeq.entrySet().iterator(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + if (entry.getValue() == listener) { + iterator.remove(); + } + } + this.pendingConfirms.remove(listener); + return result; + } + + +// ConfirmListener + + public void handleAck(long seq, boolean multiple) + throws IOException { + if (logger.isDebugEnabled()) { + logger.debug(this.toString() + " PC:Ack:" + seq + ":" + multiple); + } + this.processAck(seq, true, multiple); + } + + public void handleNack(long seq, boolean multiple) + throws IOException { + if (logger.isDebugEnabled()) { + logger.debug(this.toString() + " PC:Nack:" + seq + ":" + multiple); + } + this.processAck(seq, false, multiple); + } + + private void processAck(long seq, boolean ack, boolean multiple) { + Listener listener = this.listenerForSeq.get(seq); + if (listener != null && listener.isConfirmListener()) { + if (multiple) { + Map headMap = this.pendingConfirms.get(listener).headMap(seq + 1); + synchronized(this.pendingConfirms) { + Iterator> iterator = headMap.entrySet().iterator(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + iterator.remove(); + listener.handleConfirm(entry.getValue(), ack); + } + } + } + else { + PendingConfirm pendingConfirm = this.pendingConfirms.get(listener).remove(seq); + if (pendingConfirm != null) { + listener.handleConfirm(pendingConfirm, ack); + } + } + } else { + logger.error("No listener for seq:" + seq); + } + } + + public void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { + SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); + Assert.notNull(pendingConfirmsForListener, "Listener not registered"); + pendingConfirmsForListener.put(seq, pendingConfirm); + this.listenerForSeq.put(seq, listener); + } + +// ReturnListener + + public void handleReturn(int replyCode, + String replyText, + String exchange, + String routingKey, + AMQP.BasicProperties properties, + byte[] body) throws IOException + { + Object uuidObject = properties.getHeaders().get(RETURN_CORRELATION).toString(); + Listener listener = this.listeners.get(uuidObject); + if (listener == null || !listener.isReturnListener()) { + if (logger.isWarnEnabled()) { + logger.warn("No Listener for returned message"); + } + } + else { + listener.handleReturn(replyCode, replyText, exchange, routingKey, properties, body); + } + } + +// Object + + @Override + public int hashCode() { + return this.delegate.hashCode(); + } + + + @Override + public boolean equals(Object obj) { + if (obj == this) { + return true; + } + return this.delegate.equals(obj); + } + + @Override + public String toString() { + return "PublisherCallbackChannelImpl: " + this.delegate.toString(); + } + +} 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..0b083f55 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 +821,13 @@ ]]> + + + + + + + + + + + + + + + diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java index e7c65ae6..9621af5c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java @@ -28,6 +28,8 @@ import org.springframework.beans.factory.xml.XmlBeanFactory; import org.springframework.core.io.ClassPathResource; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import com.rabbitmq.client.Address; + /** * * @author Dave Syer @@ -79,5 +81,21 @@ public final class ConnectionFactoryParserTests { ExecutorService exec = beanFactory.getBean("execService", ExecutorService.class); assertSame(exec, executor); } + + @Test + public void testMultiHost() throws Exception { + CachingConnectionFactory connectionFactory = beanFactory.getBean("multiHost", CachingConnectionFactory.class); + assertNotNull(connectionFactory); + assertEquals(10, connectionFactory.getChannelCacheSize()); + DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory); + Address[] addresses = (Address[]) dfa.getPropertyValue("addresses"); + assertEquals(3, addresses.length); + assertEquals("host1", addresses[0].getHost()); + assertEquals(1234, addresses[0].getPort()); + assertEquals("host2", addresses[1].getHost()); + assertEquals(-1, addresses[1].getPort()); + assertEquals("host3", addresses[2].getHost()); + assertEquals(4567, addresses[2].getPort()); + } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests.java new file mode 100644 index 00000000..31735208 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests.java @@ -0,0 +1,59 @@ +/* + * 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 static org.junit.Assert.assertEquals; + +import java.util.Map; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.amqp.core.Queue; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + + +/** + * @author Gary Russell + * @since 1.0.1 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class QueueArgumentsParserTests { + + @Autowired + private ApplicationContext ctx; + + @Autowired + private Queue queue1; + + @Autowired + private Queue queue2; + + @Test + public void test() { + @SuppressWarnings("unchecked") + Map 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..60d946f2 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 @@ -13,20 +13,27 @@ package org.springframework.amqp.rabbit.config; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +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 { @@ -42,6 +49,22 @@ public final class TemplateParserTests { public void testTemplate() throws Exception { AmqpTemplate template = beanFactory.getBean("template", AmqpTemplate.class); assertNotNull(template); + DirectFieldAccessor dfa = new DirectFieldAccessor(template); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("mandatory")); + assertEquals(Boolean.FALSE, dfa.getPropertyValue("immediate")); + assertNull(dfa.getPropertyValue("returnCallback")); + assertNull(dfa.getPropertyValue("confirmCallback")); + } + + @Test + public void testTemplateWithCallbacks() throws Exception { + AmqpTemplate template = beanFactory.getBean("withCallbacks", AmqpTemplate.class); + assertNotNull(template); + DirectFieldAccessor dfa = new DirectFieldAccessor(template); + assertEquals(Boolean.TRUE, dfa.getPropertyValue("mandatory")); + assertEquals(Boolean.TRUE, dfa.getPropertyValue("immediate")); + assertNotNull(dfa.getPropertyValue("returnCallback")); + assertNotNull(dfa.getPropertyValue("confirmCallback")); } @Test @@ -50,5 +73,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..f3543586 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java @@ -0,0 +1,241 @@ +/* + * 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); + assertNotNull(reply); + + 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/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java new file mode 100644 index 00000000..bc28c161 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -0,0 +1,351 @@ +/* + * 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 + * + * 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.assertFalse; +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.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +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.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; +import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; +import org.springframework.amqp.rabbit.support.CorrelationData; +import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.beans.DirectFieldAccessor; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +public class RabbitTemplatePublisherCallbacksIntegrationTests { + + private static final String ROUTE = "test.queue"; + + private CachingConnectionFactory connectionFactory; + + private RabbitTemplate template; + + @Before + public void create() { + connectionFactory = new CachingConnectionFactory(); + // When using publisher confirms, the cache size needs to be large enough + // otherwise channels can be closed before confirms are received. + connectionFactory.setChannelCacheSize(10); + connectionFactory.setPort(BrokerTestUtils.getPort()); + connectionFactory.setPublisherConfirms(true); + template = new RabbitTemplate(connectionFactory); + } + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE); + + @Test + public void testPublisherConfirmReceived() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch.countDown(); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + assertTrue(latch.await(1000, TimeUnit.MILLISECONDS)); + assertNull(template.getUnconfirmed(0)); + } + + @Test + public void testPublisherConfirmReceivedConcurrentThreads() throws Exception { + final CountDownLatch latch = new CountDownLatch(2); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch.countDown(); + } + }); + + // Hold up the first thread so we get two channels + final CountDownLatch threadLatch = new CountDownLatch(1); + //Thread 1 + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + template.execute(new ChannelCallback() { + public Object doInRabbit(Channel channel) throws Exception { + try { + threadLatch.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + template.doSend(channel, "", ROUTE, + new SimpleMessageConverter().toMessage("message", new MessageProperties()), + new CorrelationData("def")); + return null; + } + }); + } + }); + + // Thread 2 + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + threadLatch.countDown(); + assertTrue(latch.await(5000, TimeUnit.MILLISECONDS)); + assertNull(template.getUnconfirmed(0)); + } + + @Test + public void testPublisherConfirmReceivedTwoTemplates() throws Exception { + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch1.countDown(); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + RabbitTemplate secondTemplate = new RabbitTemplate(connectionFactory); + secondTemplate.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + latch2.countDown(); + } + }); + secondTemplate.convertAndSend(ROUTE, (Object) "message", new CorrelationData("def")); + assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); + assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS)); + assertNull(template.getUnconfirmed(0)); + assertNull(secondTemplate.getUnconfirmed(0)); + } + + @Test + public void testPublisherReturns() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final List returns = new ArrayList(); + template.setReturnCallback(new ReturnCallback() { + public void returnedMessage(Message message, int replyCode, + String replyText, String exchange, String routingKey) { + returns.add(message); + latch.countDown(); + } + }); + template.setMandatory(true); + template.setImmediate(true); + template.convertAndSend(ROUTE + "junk", (Object) "message", new CorrelationData("abc")); + assertTrue(latch.await(1000, TimeUnit.MILLISECONDS)); + assertEquals(1, returns.size()); + Message message = returns.get(0); + assertEquals("message", new String(message.getBody(), "utf-8")); + } + + @Test + public void testPublisherConfirmNotReceived() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(new PublisherCallbackChannelImpl(mockChannel)); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final AtomicBoolean confirmed = new AtomicBoolean(); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + confirmed.set(true); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + Thread.sleep(5); + Collection unconfirmed = template.getUnconfirmed(0); + assertEquals(1, unconfirmed.size()); + assertEquals("abc", unconfirmed.iterator().next().getId()); + assertFalse(confirmed.get()); + } + + @Test + public void testPublisherConfirmNotReceivedMultiThreads() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + PublisherCallbackChannelImpl channel1 = new PublisherCallbackChannelImpl(mockChannel); + PublisherCallbackChannelImpl channel2 = new PublisherCallbackChannelImpl(mockChannel); + when(mockConnection.createChannel()).thenReturn(channel1).thenReturn(channel2); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final AtomicBoolean confirmed = new AtomicBoolean(); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + confirmed.set(true); + } + }); + + // Hold up the first thread so we get two channels + final CountDownLatch threadLatch = new CountDownLatch(1); + final CountDownLatch threadSentLatch = new CountDownLatch(1); + //Thread 1 + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + template.execute(new ChannelCallback() { + public Object doInRabbit(Channel channel) throws Exception { + try { + threadLatch.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + template.doSend(channel, "", ROUTE, + new SimpleMessageConverter().toMessage("message", new MessageProperties()), + new CorrelationData("def")); + threadSentLatch.countDown(); + return null; + } + }); + } + }); + + // Thread 2 + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + threadLatch.countDown(); + assertTrue(threadSentLatch.await(5, TimeUnit.SECONDS)); + Thread.sleep(5); + Collection unconfirmed = template.getUnconfirmed(0); + assertEquals(2, unconfirmed.size()); + Set ids = new HashSet(); + Iterator iterator = unconfirmed.iterator(); + ids.add(iterator.next().getId()); + ids.add(iterator.next().getId()); + assertTrue(ids.remove("abc")); + assertTrue(ids.remove("def")); + assertFalse(confirmed.get()); + DirectFieldAccessor dfa = new DirectFieldAccessor(template); + Map pendingConfirms = (Map) dfa.getPropertyValue("pendingConfirms"); + assertEquals(2, pendingConfirms.size()); + channel1.close(); + assertEquals(1, pendingConfirms.size()); + channel2.close(); + assertEquals(0, pendingConfirms.size()); + } + + @Test + public void testPublisherConfirmNotReceivedAged() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(new PublisherCallbackChannelImpl(mockChannel)); + + final AtomicInteger count = new AtomicInteger(); + doAnswer(new Answer(){ + public Object answer(InvocationOnMock invocation) throws Throwable { + return count.incrementAndGet(); + }}).when(mockChannel).getNextPublishSeqNo(); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final AtomicBoolean confirmed = new AtomicBoolean(); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + confirmed.set(true); + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + Thread.sleep(100); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("def")); + Collection unconfirmed = template.getUnconfirmed(50); + assertEquals(1, unconfirmed.size()); + assertEquals("abc", unconfirmed.iterator().next().getId()); + assertFalse(confirmed.get()); + Thread.sleep(100); + unconfirmed = template.getUnconfirmed(50); + assertEquals(1, unconfirmed.size()); + assertEquals("def", unconfirmed.iterator().next().getId()); + assertFalse(confirmed.get()); + } + + @Test + public void testPublisherConfirmMultiple() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + PublisherCallbackChannelImpl callbackChannel = new PublisherCallbackChannelImpl(mockChannel); + when(mockConnection.createChannel()).thenReturn(callbackChannel); + + final AtomicInteger count = new AtomicInteger(); + doAnswer(new Answer(){ + public Object answer(InvocationOnMock invocation) throws Throwable { + return count.incrementAndGet(); + }}).when(mockChannel).getNextPublishSeqNo(); + + final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory)); + + final List confirms = new ArrayList(); + final CountDownLatch latch = new CountDownLatch(2); + template.setConfirmCallback(new ConfirmCallback() { + + public void confirm(CorrelationData correlationData, boolean ack) { + if (ack) { + confirms.add(correlationData.getId()); + latch.countDown(); + } + } + }); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("def")); + callbackChannel.handleAck(2, true); + assertTrue(latch.await(1000, TimeUnit.MILLISECONDS)); + Collection unconfirmed = template.getUnconfirmed(0); + assertNull(unconfirmed); + } +} 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..0a4b1a1b 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,27 @@ 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