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

Merge pull request #26 from garyrussell/AMQP-213

Browse Source 
* AMQP-213c:
  AMQP-213 Publisher Confirms/Returns
  AMQP-206 Support Rabbit HA
This commit is contained in:
Oleg Zhurakousky
2012-05-10 13:41:56 -04:00
parent a356eb24bb 5944e34b6e
commit 6b8dd72b98
30 changed files with 2679 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

37
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -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<? extends ConnectionListener> 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 {

381
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,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<String, LinkedBlockingQueue<Message>> replyHolder = new ConcurrentHashMap<String, LinkedBlockingQueue<Message>>();
private volatile ConfirmCallback confirmCallback;
private volatile ReturnCallback returnCallback;
private final Map<Object, SortedMap<Long, PendingConfirm>> pendingConfirms = new ConcurrentHashMap<Object, SortedMap<Long, PendingConfirm>>();
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<CorrelationData> getUnconfirmed(long age) {
Set<CorrelationData> unconfirmed = new HashSet<CorrelationData>();
synchronized (this.pendingConfirms) {
long threshold = System.currentTimeMillis() - age;
for (Entry<Object, SortedMap<Long, PendingConfirm>> channelPendingConfirmEntry : this.pendingConfirms.entrySet()) {
SortedMap<Long, PendingConfirm> channelPendingConfirms = channelPendingConfirmEntry.getValue();
Iterator<Entry<Long, PendingConfirm>> 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<Object>() {
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<Message>() {
public Message doInRabbit(Channel channel) throws Exception {
final SynchronousQueue<Message> replyHandoff = new SynchronousQueue<Message>();
@@ -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<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, null);
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();
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<Long, PendingConfirm> 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<Long, PendingConfirm> 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<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;
}
}
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);
}
}

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

45
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java Normal file
View File

@@ -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 + "]";
}
}

55
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java Normal file
View File

@@ -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 + "]";
}
}

104
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java Normal file
View File

@@ -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 <b>NOT</b> 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<Long, PendingConfirm> 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<Long, PendingConfirm> unconfirmed);
/**
* Returns the UUID used to identify this Listener for returns.
* @return A string representation of the UUID.
*/
String getUUID();
boolean isConfirmListener();
boolean isReturnListener();
}
}

559
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java Normal file
View File

@@ -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<String, Listener> listeners = new ConcurrentHashMap<String, Listener>();
private final Map<Listener, SortedMap<Long, PendingConfirm>> pendingConfirms
= new ConcurrentHashMap<PublisherCallbackChannel.Listener, SortedMap<Long,PendingConfirm>>();
private final Map<Long, Listener> listenerForSeq = new ConcurrentHashMap<Long, Listener>();
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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<Listener, SortedMap<Long, PendingConfirm>> entry : this.pendingConfirms.entrySet()) {
Listener listener = entry.getKey();
listener.removePendingConfirmsReference(this, entry.getValue());
}
this.pendingConfirms.clear();
this.listenerForSeq.clear();
}
public synchronized SortedMap<Long, PendingConfirm> 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<Long, PendingConfirm>()));
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<Entry<Long, Listener>> iterator = this.listenerForSeq.entrySet().iterator();
while (iterator.hasNext()) {
Entry<Long, Listener> 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<Long, PendingConfirm> headMap = this.pendingConfirms.get(listener).headMap(seq + 1);
synchronized(this.pendingConfirms) {
Iterator<Entry<Long, PendingConfirm>> iterator = headMap.entrySet().iterator();
while (iterator.hasNext()) {
Entry<Long, PendingConfirm> 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<Long, PendingConfirm> 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();
}
}

361
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,63 @@
</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:attribute name="mandatory" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When 'true' sets the mandatory flag on basic.publish; only applies if
a 'return-callback' is provided.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="immediate" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When 'true' sets the immediate flag on basic.publish; only applies if
a 'return-callback' is provided.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="return-callback" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to an implementation of RabbitTemplate.ReturnCallback - invoked if
a return is received for a message published with mandatory or immediate set
that couldn't be delivered according to the semantics of those options.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.RabbitTemplate.ReturnCallback" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-callback" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to an implementation of RabbitTemplate.ConfirmCallback - invoked if
a confirm (ack or nack) return is received for a published message.
Requires a 'connection-factory' that has 'publisher-confirms' set to 'true'.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.amqp.core.RabbitTemplate.ConfirmCallback" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -733,6 +821,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[
@@ -788,6 +883,20 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="publisher-confirms" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When true, channels on connections created by this factory support publisher confirms.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="publisher-returns" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When true, channels on connections created by this factory support publisher confirms.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

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

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

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

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

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

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

@@ -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<Object>() {
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<Message> returns = new ArrayList<Message>();
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<CorrelationData> 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<Object>() {
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<CorrelationData> unconfirmed = template.getUnconfirmed(0);
assertEquals(2, unconfirmed.size());
Set<String> ids = new HashSet<String>();
Iterator<CorrelationData> 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<Object>(){
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<CorrelationData> 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<Object>(){
public Object answer(InvocationOnMock invocation) throws Throwable {
return count.incrementAndGet();
}}).when(mockChannel).getNextPublishSeqNo();
final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory));
final List<String> confirms = new ArrayList<String>();
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<CorrelationData> unconfirmed = template.getUnconfirmed(0);
assertNull(unconfirmed);
}
}

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>

24
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,27 @@
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>
<rabbit:template id="withCallbacks" connection-factory="connectionFactory"
mandatory="true" immediate="true" return-callback="rcb" confirm-callback="ccb" />
<beans:bean id="rcb" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.amqp.rabbit.core.RabbitTemplate$ReturnCallback" />
</beans:bean>
<beans:bean id="ccb" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.springframework.amqp.rabbit.core.RabbitTemplate$ConfirmCallback" />
</beans:bean>
</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