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

AMQP-206 Support Rabbit HA

Browse Source 
- Configure a list of addresses in ConnectionFactory
- Allow arguments (e.g. HA) for temporary reply queue creation

AMQP-206 Add queue-arguments Attribute

<queue/> now supports either <queue-arguments /> element
or a queue-arguments attribute which is a reference to
a <queue-arguments /> bean.

<queue-arguments /> is promoted to a top-level bean,
allowing for re-use, such as...

    <rabbit:queue-arguments id="haArgs">
	<entry key="x-ha-policy" value="all" />
    </rabbit:queue-arguments>

    <rabbit:queue name="si.test.queue" queue-arguments="haArgs" />

AMQP-206 Fix Parser Tests

AMQP-206 ReConnect To Mirrored Queue

If a consumer is connected to a mirror (slave), and the
master dies, the queue is moved. The consumer is given a
cancel notification, indicating we need to reconnect.

However, the move might not be complete, so we retry the
queue declaration a number of times before failing.

AMQP-206 Polishing

Cleanup retry counter name; trap exception on cancel of
temporary reply queue.

AMQP-206 Add reply-queue to RabbitTemplate

Allow specification of an explicit queue for replies. Adds
a correlation header to sendAndReceive messages (spring_reply_correlation)
which is used to correlate replies on the reply queue.

The queue is supplied to the template via the reply-queue
attribute; the template must be added as a <listener/> to a
<listener-container/> for the same queue. For example...

    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" reply-queue="si.test.replies" />

    <rabbit:listener-container id="replyContainer" connection-factory="connectionFactory" task-executor="reply-exec">
    	<rabbit:listener ref="amqpTemplate" queues="si.test.replies"/>
    </rabbit:listener-container>

    <rabbit:queue name="si.test.replies" queue-arguments="haArgs" />

AMQP-206 Stack Reply Correlation

Stacks (pushes/pops) correlation when using a fixed queue for
replies in sendAndReceive methods.

Now supports

templatsendAndReceive()->listener->templateSendAndReceive()->
   listener->templateSendAndReceive->...

and the replies get correlated in each template appropriately.

AMQP-206 Polishing

Add exclusive test for addresses attribute; add parser test.

AMQP-206 Polishing

Improve configuration of reply-queue listener on rabbit template.
Previously a disjoint listener was required. Now the listener
is a child element of the template itself.

	<rabbit:template id="withReplyQ" connection-factory="connectionFactory" reply-queue="reply.queue">
		<rabbit:reply-listener />
	</rabbit:template>

The reply-listener is a full-blown listener container, but uses the
template's connection factory and so that attribute, as well as the
message-converter is disallowed at run-time.

AMQP-206 Polishing

Remove disallowed attributes from <retry-listener/> child element
in schema.

AMQP-206 Reference Documentation

Update docs with addresses attribute on ConnectionFactory and
reply-queue on template.
This commit is contained in:
Gary Russell
2012-01-11 11:45:19 -05:00
committed by Oleg Zhurakousky
parent a356eb24bb
commit fd696bfaf9
24 changed files with 1227 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java
View File

@@ -33,6 +33,8 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
private static final String PORT_ATTRIBUTE = "port";
private static final String ADDRESSES = "addresses";
private static final String VIRTUAL_HOST_ATTRIBUTE = "virtual-host";
private static final String USER_ATTRIBUTE = "username";
@@ -58,7 +60,11 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
if (element.hasAttribute(ADDRESSES) &&
(element.hasAttribute(HOST_ATTRIBUTE) || element.hasAttribute(PORT_ATTRIBUTE))) {
parserContext.getReaderContext().error("If the 'addresses' attribute is provided, a connection " +
"factory can not have 'host' or 'port' attributes.", element);
}
NamespaceUtils.addConstructorArgParentRefIfAttributeDefined(builder, element, CONNECTION_FACTORY_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, CHANNEL_CACHE_SIZE_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, HOST_ATTRIBUTE);
@@ -67,6 +73,7 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
NamespaceUtils.setValueIfAttributeDefined(builder, element, PASSWORD_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, VIRTUAL_HOST_ATTRIBUTE);
NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, ADDRESSES);
}
}

132
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerParser.java
View File

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

46
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueArgumentsParser.java Normal file
View File

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

17
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java
View File

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

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java
View File

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

162
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java Normal file
View File

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

71
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2010-2011 the original author or authors.
* Copyright 2010-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -14,14 +14,19 @@
package org.springframework.amqp.rabbit.config;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
/**
* @author Dave Syer
* @author Gary Russell
*/
class TemplateParser extends AbstractSingleBeanDefinitionParser {
@@ -41,6 +46,10 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser {
private static final String CHANNEL_TRANSACTED_ATTRIBUTE = "channel-transacted";
private static final String REPLY_QUEUE_ATTRIBUTE = "reply-queue";
private static final String LISTENER_ELEMENT = "reply-listener";
@Override
protected Class<?> getBeanClass(Element element) {
return RabbitTemplate.class;
@@ -53,7 +62,7 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser {
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
return false;
}
@Override
@@ -77,7 +86,65 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser {
NamespaceUtils.setValueIfAttributeDefined(builder, element, REPLY_TIMEOUT_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, ENCODING_ATTRIBUTE);
NamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_CONVERTER_ATTRIBUTE);
NamespaceUtils.setReferenceIfAttributeDefined(builder, element, REPLY_QUEUE_ATTRIBUTE);
BeanDefinition replyContainer = null;
Element childElement = getChildElement(element, parserContext);
if (childElement != null) {
replyContainer = parseListener(childElement, element,
parserContext);
if (replyContainer != null) {
replyContainer.getPropertyValues().add("messageListener",
new RuntimeBeanReference(element.getAttribute(ID_ATTRIBUTE)));
String replyContainerName = element.getAttribute(ID_ATTRIBUTE) + ".replyListener";
parserContext.getRegistry().registerBeanDefinition(replyContainerName, replyContainer);
}
}
if (replyContainer == null && element.hasAttribute(REPLY_QUEUE_ATTRIBUTE)) {
parserContext.getReaderContext().error(
"For template '" + element.getAttribute(ID_ATTRIBUTE)
+ "', when specifying a reply-queue, "
+ "a <reply-listener/> element is required",
element);
}
else if (replyContainer != null && !element.hasAttribute(REPLY_QUEUE_ATTRIBUTE)) {
parserContext.getReaderContext().error(
"For template '" + element.getAttribute(ID_ATTRIBUTE)
+ "', a <reply-listener/> element is not allowed if no " +
"'reply-queue' is supplied",
element);
}
}
private Element getChildElement(Element element,
ParserContext parserContext) {
Element childElement = null;
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE) {
String localName = parserContext.getDelegate().getLocalName(child);
if (LISTENER_ELEMENT.equals(localName)) {
childElement = (Element) child;
}
}
}
return childElement;
}
private BeanDefinition parseListener(Element childElement, Element element,
ParserContext parserContext) {
if (getChildElement(childElement, parserContext) != null) {
parserContext.getReaderContext().error("<reply-listener/> is not allowed any child elements.", element);
}
BeanDefinition replyContainer = RabbitNamespaceUtils.parseContainer(childElement, parserContext);
if (replyContainer != null) {
replyContainer.getPropertyValues().add(
"connectionFactory",
new RuntimeBeanReference(element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE)));
}
replyContainer.getPropertyValues().add("queues", element.getAttribute(REPLY_QUEUE_ATTRIBUTE));
return replyContainer;
}
}

22
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java
View File

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

160
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -14,15 +14,20 @@
package org.springframework.amqp.rabbit.core;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.AmqpIllegalStateException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
@@ -33,6 +38,7 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.Queue.DeclareOk;
@@ -73,9 +79,10 @@ import com.rabbitmq.client.GetResponse;
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
* @author Gary Russell
* @since 1.0
*/
public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, MessageListener {
private static final String DEFAULT_EXCHANGE = ""; // alias for amq.direct default exchange
@@ -98,7 +105,15 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
private String encoding = DEFAULT_ENCODING;
private volatile String encoding = DEFAULT_ENCODING;
private volatile Queue replyQueue;
private final Map<String, LinkedBlockingQueue<Message>> replyHolder = new ConcurrentHashMap<String, LinkedBlockingQueue<Message>>();
public static final String STACKED_CORRELATION_HEADER = "spring_reply_correlation";
public static final String STACKED_REPLY_TO_HEADER = "spring_reply_to";
/**
* Convenient constructor for use with setter injection. Don't forget to set the connection factory.
@@ -164,6 +179,17 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
this.encoding = encoding;
}
/**
* A queue for replies; if not provided, a temporary exclusive, auto-delete queue will
* be used for each reply.
*
* @param replyQueue the replyQueue to set
*/
public void setReplyQueue(Queue replyQueue) {
this.replyQueue = replyQueue;
}
/**
* Specify the timeout in milliseconds to be used when waiting for a reply Message when using one of the
* sendAndReceive methods. The default value is defined as {@link #DEFAULT_REPLY_TIMEOUT}. A negative value
@@ -358,6 +384,15 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
* @return the message that is received in reply
*/
protected Message doSendAndReceive(final String exchange, final String routingKey, final Message message) {
if (this.replyQueue == null) {
return doSendAndReceiveWithTemporary(exchange, routingKey, message);
}
else {
return doSendAndReceiveWithFixed(exchange, routingKey, message);
}
}
protected Message doSendAndReceiveWithTemporary(final String exchange, final String routingKey, final Message message) {
Message replyMessage = this.execute(new ChannelCallback<Message>() {
public Message doInRabbit(Channel channel) throws Exception {
final SynchronousQueue<Message> replyHandoff = new SynchronousQueue<Message>();
@@ -400,6 +435,50 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
return replyMessage;
}
protected Message doSendAndReceiveWithFixed(final String exchange, final String routingKey, final Message message) {
Message replyMessage = this.execute(new ChannelCallback<Message>() {
public Message doInRabbit(Channel channel) throws Exception {
final LinkedBlockingQueue<Message> replyHandoff = new LinkedBlockingQueue<Message>();
String messageTag = UUID.randomUUID().toString();
RabbitTemplate.this.replyHolder.put(messageTag, replyHandoff);
String replyTo = message.getMessageProperties().getReplyTo();
if (StringUtils.hasLength(replyTo) && logger.isDebugEnabled()) {
logger.debug("Dropping replyTo header:" + replyTo
+ " in favor of template's configured reply-queue:"
+ RabbitTemplate.this.replyQueue.getName());
}
String springReplyTo = (String) message.getMessageProperties()
.getHeaders().get(STACKED_REPLY_TO_HEADER);
message.getMessageProperties().setHeader(
STACKED_REPLY_TO_HEADER,
pushHeaderValue(replyTo,
springReplyTo));
message.getMessageProperties().setReplyTo(RabbitTemplate.this.replyQueue.getName());
String correlation = (String) message.getMessageProperties()
.getHeaders().get(STACKED_CORRELATION_HEADER);
if (StringUtils.hasLength(correlation)) {
message.getMessageProperties().setHeader(
STACKED_CORRELATION_HEADER,
pushHeaderValue(messageTag, correlation));
} else {
message.getMessageProperties().setHeader(
"spring_reply_correlation", messageTag);
}
if (logger.isDebugEnabled()) {
logger.debug("Sending message with tag " + messageTag);
}
doSend(channel, exchange, routingKey, message);
Message reply = (replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout,
TimeUnit.MILLISECONDS);
RabbitTemplate.this.replyHolder.remove(messageTag);
return reply;
}
});
return replyMessage;
}
public <T> T execute(ChannelCallback<T> action) {
Assert.notNull(action, "Callback object must not be null");
RabbitResourceHolder resourceHolder = getTransactionalResourceHolder();
@@ -483,4 +562,79 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations {
return name;
}
public void onMessage(Message message) {
String messageTag = (String) message.getMessageProperties()
.getHeaders().get(STACKED_CORRELATION_HEADER);
if (messageTag == null) {
logger.error("No correlation header in reply");
return;
}
PoppedHeader poppedHeaderValue = popHeaderValue(messageTag);
messageTag = poppedHeaderValue.getPoppedValue();
message.getMessageProperties().setHeader(STACKED_CORRELATION_HEADER,
poppedHeaderValue.getNewValue());
String springReplyTo = (String) message.getMessageProperties()
.getHeaders().get(STACKED_REPLY_TO_HEADER);
if (springReplyTo != null) {
poppedHeaderValue = popHeaderValue(springReplyTo);
springReplyTo = poppedHeaderValue.getNewValue();
message.getMessageProperties().setHeader(STACKED_REPLY_TO_HEADER, springReplyTo);
message.getMessageProperties().setReplyTo(null);
}
LinkedBlockingQueue<Message> 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;
}
}
}

74
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

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

30
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConsumerCancelledException.java Normal file
View File

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

303
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd
View File

@@ -56,6 +56,18 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="queue-arguments" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to a <queue-arguments/> element.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.Map" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -338,6 +350,7 @@
<xsd:complexType name="mapType">
<xsd:complexContent>
<xsd:extension base="beans:mapType">
<xsd:attribute name="id" type="xsd:string" />
<xsd:attribute name="ref" use="optional">
<xsd:annotation>
<xsd:documentation source="java:java.util.Map"><![CDATA[
@@ -368,153 +381,160 @@
</xsd:appinfo>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element name="listener" type="listenerType" minOccurs="0" maxOccurs="unbounded" />
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
<xsd:complexContent>
<xsd:extension base="listenerContainerBaseType">
<xsd:sequence>
<xsd:element name="listener" type="listenerType" minOccurs="0" maxOccurs="unbounded" />
</xsd:sequence>
<xsd:attribute name="connection-factory" type="xsd:string" default="rabbitConnectionFactory">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to the org.springframework.amqp.rabbit.connection.ConnectionFactory.
Default referenced bean name is "rabbitConnectionFactory".
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.rabbit.connection.ConnectionFactory" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
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.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.support.converter.MessageConverter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="listenerContainerBaseType">
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Optional bean id for the container.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="connection-factory" type="xsd:string" default="rabbitConnectionFactory">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to the org.springframework.amqp.rabbit.connection.ConnectionFactory.
Default referenced bean name is "rabbitConnectionFactory".
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.rabbit.connection.ConnectionFactory" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to a Spring TaskExecutor (or standard JDK 1.5 Executor) for executing
listener invokers. Default is a SimpleAsyncTaskExecutor, using internally managed threads.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.concurrent.Executor" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
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.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.support.converter.MessageConverter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-handler" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.concurrent.Executor" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-handler" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to an ErrorHandler strategy for handling any uncaught Exceptions
that may occur during the execution of the MessageListener.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.util.ErrorHandler" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="acknowledge" default="auto">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.util.ErrorHandler" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="acknowledge" default="auto">
<xsd:annotation>
<xsd:documentation><![CDATA[
The acknowledge mode: "auto", "manual", or "none".
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="auto" />
<xsd:enumeration value="manual" />
<xsd:enumeration value="none" />
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="transaction-manager" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="auto" />
<xsd:enumeration value="manual" />
<xsd:enumeration value="none" />
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="transaction-manager" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to an external PlatformTransactionManager.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.transaction.PlatformTransactionManager" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="concurrency" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.transaction.PlatformTransactionManager" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="concurrency" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The number of concurrent consumers to start for each listener.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="prefetch" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="prefetch" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Tells the broker how many messages to send to each consumer in a single request. Often this can be set quite high
to improve throughput. It should be greater than or equal to the transaction size.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="transaction-size" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="transaction-size" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Tells the container how many messages to process in a single transaction (if the channel is transactional). For
best results it should be less than or equal to the prefetch count.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="requeue-rejected" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="requeue-rejected" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Tells the container the default requeue behavior when rejecting messages. Default is 'true' meaning messages
will be requeued, unless the listener signals not to by throwing an AmqpRejectAndDontRequeueException. When
set to false, messages will never be requeued.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="phase" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="phase" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
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.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Flag to indicate that the container should start up automatically when the enclosing context is refreshed. Default true.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="advice-chain" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="advice-chain" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to a chain of AOP advice to be applied to the listener.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="listenerType">
<xsd:attribute name="id" type="xsd:string">
@@ -629,6 +649,17 @@
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element name="reply-listener" type="listenerContainerBaseType"
minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
A <listener-container/> used to receive asynchronous replies on the reply-channel the
<listener/> child element is disallowed because the template itself is the listener.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -702,6 +733,19 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-queue" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to a <queue/> for replies; optional; if not supplied, methods expecting replies
will use a temporary, exclusive, auto-delete queue.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.Queue" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -733,6 +777,13 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="addresses" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
List of addresses; e.g. host1,host2:4567,host3 - overrides host/port if supplied.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="username" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[

18
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java
View File

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

59
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests.java Normal file
View File

@@ -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<String, Object> args = (Map<String, Object>) ctx.getBean("args");
assertEquals("bar", args.get("foo"));
assertEquals("qux", queue1.getArguments().get("baz"));
assertEquals("bar", queue2.getArguments().get("foo"));
}
}

7
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserTests.java
View File

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

22
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java
View File

@@ -14,19 +14,24 @@
package org.springframework.amqp.rabbit.config;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import org.junit.Before;
import org.junit.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.SerializerMessageConverter;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.xml.XmlBeanFactory;
import org.springframework.core.io.ClassPathResource;
/**
*
* @author Dave Syer
* @author Gary Russell
*
*/
public final class TemplateParserTests {
@@ -50,5 +55,20 @@ public final class TemplateParserTests {
assertNotNull(template);
assertTrue(template.getMessageConverter() instanceof SerializerMessageConverter);
}
@Test
public void testWithReplyQ() throws Exception {
RabbitTemplate template = beanFactory.getBean("withReplyQ", RabbitTemplate.class);
assertNotNull(template);
DirectFieldAccessor dfa = new DirectFieldAccessor(template);
Queue queue = (Queue) dfa.getPropertyValue("replyQueue");
assertNotNull(queue);
Queue queueBean = beanFactory.getBean("reply.queue", Queue.class);
assertSame(queueBean, queue);
SimpleMessageListenerContainer container = beanFactory.getBean("withReplyQ.replyListener", SimpleMessageListenerContainer.class);
assertNotNull(container);
dfa = new DirectFieldAccessor(container);
assertSame(template, dfa.getPropertyValue("messageListener"));
}
}

240
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java Normal file
View File

@@ -0,0 +1,240 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.core;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.util.ReflectionUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* @author Gary Russell
* @since 1.0.1
*
*/
public class RabbitTemplateHeaderTests {
@Test
public void testPushPop() throws Exception {
RabbitTemplate template = new RabbitTemplate();
Method pushHeader = RabbitTemplate.class.getDeclaredMethod("pushHeaderValue", String.class, String.class);
ReflectionUtils.makeAccessible(pushHeader);
String header = (String) ReflectionUtils.invokeMethod(pushHeader, template, "a", null);
assertEquals("a", header);
header = (String) ReflectionUtils.invokeMethod(pushHeader, template, "b", header);
assertEquals("b:a", header);
header = (String) ReflectionUtils.invokeMethod(pushHeader, template, "c", header);
assertEquals("c:b:a", header);
Method popHeader = RabbitTemplate.class.getDeclaredMethod("popHeaderValue", String.class);
ReflectionUtils.makeAccessible(popHeader);
Object poppedHeader = ReflectionUtils.invokeMethod(popHeader, template, header);
DirectFieldAccessor dfa = new DirectFieldAccessor(poppedHeader);
assertEquals("c", dfa.getPropertyValue("poppedValue"));
poppedHeader = ReflectionUtils.invokeMethod(popHeader, template, dfa.getPropertyValue("newValue"));
dfa = new DirectFieldAccessor(poppedHeader);
assertEquals("b", dfa.getPropertyValue("poppedValue"));
poppedHeader = ReflectionUtils.invokeMethod(popHeader, template, dfa.getPropertyValue("newValue"));
dfa = new DirectFieldAccessor(poppedHeader);
assertEquals("a", dfa.getPropertyValue("poppedValue"));
assertNull(dfa.getPropertyValue("newValue"));
}
@Test
public void testReplyToOneDeep() throws Exception {
ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
when(mockConnection.createChannel()).thenReturn(mockChannel);
final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory));
Queue replyQueue = new Queue("new.replyTo");
template.setReplyQueue(replyQueue);
MessageProperties messageProperties = new MessageProperties();
messageProperties.setReplyTo("replyTo1");
Message message = new Message("Hello, world!".getBytes(), messageProperties);
final List<BasicProperties> props = new ArrayList<BasicProperties>();
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
BasicProperties basicProps = (BasicProperties) invocation.getArguments()[4];
props.add(basicProps);
MessageProperties springProps = new DefaultMessagePropertiesConverter()
.toMessageProperties(basicProps, null, "UTF-8");
Message replyMessage = new Message("!dlrow olleH".getBytes(), springProps);
template.onMessage(replyMessage);
return null;
}}
).when(mockChannel).basicPublish(Mockito.any(String.class),
Mockito.any(String.class), Mockito.anyBoolean(),
Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
Message reply = template.sendAndReceive(message);
assertEquals(1, props.size());
BasicProperties basicProperties = props.get(0);
assertEquals("new.replyTo", basicProperties.getReplyTo());
assertEquals("replyTo1", basicProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER));
assertNotNull(basicProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER));
}
@Test
public void testReplyToTwoDeep() throws Exception {
ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
when(mockConnection.createChannel()).thenReturn(mockChannel);
final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory));
Queue replyQueue = new Queue("new.replyTo");
template.setReplyQueue(replyQueue);
MessageProperties messageProperties = new MessageProperties();
messageProperties.setReplyTo("replyTo2");
messageProperties.setHeader(RabbitTemplate.STACKED_REPLY_TO_HEADER, "replyTo1");
messageProperties.setHeader(RabbitTemplate.STACKED_CORRELATION_HEADER, "a");
Message message = new Message("Hello, world!".getBytes(), messageProperties);
final List<BasicProperties> props = new ArrayList<BasicProperties>();
doAnswer(new Answer<Object>() {
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<BasicProperties> props = new ArrayList<BasicProperties>();
final AtomicInteger count = new AtomicInteger();
final List<String> nestedReplyTo = new ArrayList<String>();
final List<String> nestedReplyStack = new ArrayList<String>();
final List<String> nestedCorrelation = new ArrayList<String>();
doAnswer(new Answer<Object>() {
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"));
}
}

3
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml
View File

@@ -29,4 +29,7 @@
<bean id="execService" class="java.util.concurrent.Executors" factory-method="newSingleThreadExecutor" />
<rabbit:connection-factory id="multiHost" virtual-host="/bar" addresses="host1:1234,host2,host3:4567"
channel-cache-size="10" username="user" password="password" />
</beans>

20
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueArgumentsParserTests-context.xml Normal file
View File

@@ -0,0 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<rabbit:queue-arguments id="args">
<entry key="foo" value="bar" />
</rabbit:queue-arguments>
<rabbit:queue id="queue1">
<rabbit:queue-arguments>
<entry key="baz" value="qux" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue id="queue2" queue-arguments="args" />
</beans>

6
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserPlaceholderTests-context.xml
View File

@@ -38,4 +38,10 @@
<rabbit:queue id="anonymousOverride" durable="false" auto-delete="true" exclusive="true" />
<rabbit:queue-arguments id="topLevelArgs">
<beans:entry key="baz" value="qux" />
</rabbit:queue-arguments>
<rabbit:queue id="referencedArguments" queue-arguments="topLevelArgs" />
</beans>

6
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/QueueParserTests-context.xml
View File

@@ -28,4 +28,10 @@
<rabbit:queue id="anonymousOverride" durable="false" auto-delete="true" exclusive="true" />
<rabbit:queue-arguments id="topLevelArgs">
<beans:entry key="baz" value="qux" />
</rabbit:queue-arguments>
<rabbit:queue id="referencedArguments" queue-arguments="topLevelArgs" />
</beans>

14
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml
View File

@@ -1,7 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<rabbit:template id="template" connection-factory="connectionFactory" />
@@ -10,7 +10,17 @@
encoding="UTF-8" exchange="foo" queue="bar" routing-key="spam" message-converter="converter" reply-timeout="1000" />
<rabbit:connection-factory id="connectionFactory" />
<bean id="converter" class="org.springframework.amqp.support.converter.SerializerMessageConverter"/>
<rabbit:template id="withReplyQ" connection-factory="connectionFactory" reply-queue="reply.queue">
<rabbit:reply-listener />
</rabbit:template>
<rabbit:queue name="reply.queue" queue-arguments="args" />
<rabbit:queue-arguments id="args">
<entry key="foo" value="bar" />
</rabbit:queue-arguments>
</beans>

87
src/docbkx/amqp.xml
View File

@@ -296,6 +296,17 @@ Connection connection = connectionFactory.createConnection();]]></programlisting
<programlisting language="xml"><![CDATA[<rabbit:connection-factory
id="connectionFactory" channel-cache-size="25"/>]]></programlisting>
</para>
<para>
Host and port attributes can be provided using the namespace
<programlisting language="xml"><![CDATA[<rabbit:connection-factory
id="connectionFactory" host="somehost" port="5672" />]]></programlisting>
</para>
<para>
Alternatively, if running in a clustered environment, use the addresses
attribute.
<programlisting language="xml"><![CDATA[<rabbit:connection-factory
id="connectionFactory" addresses="host1:5672,host2:5672" />]]></programlisting>
</para>
</section>
<section id="amqp-template">
@@ -685,6 +696,23 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]></programlist
<para>Similar request/reply methods are also available where the <classname>MessageConverter</classname>
is applied to both the request and reply. Those methods are named <methodname>convertSendAndReceive</methodname>.
See the Javadoc of <classname>AmqpTemplate</classname> for more detail.</para>
<para>
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 &lt;reply-listener/&gt; sub element. This element provides a listener container for the
reply queue, with the template being the listener. All of the <xref linkend="containerAttributes" /> attributes
allowed on a &lt;listener-container/&gt; are allowed on the element, except for connection-factory and
message-converter, which are inherited from the template's configuration.
<programlisting language="xml"><![CDATA[<rabbit:template id="amqpTemplate"
connection-factory="connectionFactory" reply-queue="replies">
<rabbit:reply-listener />
</rabbit:template>
]]></programlisting>
</para>
<para>
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).
</para>
</section>
<section id="broker-configuration">
@@ -997,14 +1025,18 @@ public class ExampleExternalTransactionAmqpConfiguration {
</section>
</section>
<section>
<title>Message Listener Container Features</title>
<section id="containerAttributes">
<title>Message Listener Container Configuration</title>
<para>There are quite a few options for configuring a
<classname>SimpleMessageListenerContainer</classname> related to
transactions and quality of service, and some of them interact with each
other.</para>
<para>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'.</para>
<para><table>
<title>Configuration options for a message listener container</title>
@@ -1105,6 +1137,17 @@ public class ExampleExternalTransactionAmqpConfiguration {
broker is ready.</entry>
</row>
<row>
<entry>phase</entry>
<entry>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.</entry>
</row>
<row>
<entry>adviceChain</entry>
@@ -1115,6 +1158,46 @@ public class ExampleExternalTransactionAmqpConfiguration {
the <classname>CachingConnectionFactory</classname>, as long as
the broker is still alive.</entry>
</row>
<row>
<entry>taskExecutor</entry>
<entry>A reference to a Spring TaskExecutor (or standard JDK 1.5+
Executor) for executing listener invokers. Default is a
SimpleAsyncTaskExecutor, using internally managed threads.</entry>
</row>
<row>
<entry>errorHandler</entry>
<entry>A reference to an ErrorHandler strategy for handling any
uncaught Exceptions that may occur during the execution of the
MessageListener.</entry>
</row>
<row>
<entry>concurrency</entry>
<entry>The number of concurrent consumers to start for each
listener.</entry>
</row>
<row>
<entry>connectionFactory</entry>
<entry>A reference to the connectionFactory; when configuring
using the XML namespace, the default referenced bean name
is "rabbitConnectionFactory".</entry>
</row>
<row>
<entry>messageConverter</entry>
<entry>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.</entry>
</row>
</tbody>
</tgroup>
</table></para>

6
src/docbkx/index.xml
View File

@@ -23,10 +23,14 @@
<firstname>Dave</firstname>
<surname>Syer</surname>
</author>
<author>
<firstname>Gary</firstname>
<surname>Russell</surname>
</author>
</authorgroup>
<legalnotice>
<para>Copyright &#xA9; 2010-2011</para>
<para>Copyright &#xA9; 2010-2012</para>
<para>
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