AMQP-213 Publisher Confirms/Returns

Initial Commit

Add support for multiple acks in one call.

AMQP-213 Polishing

Move confirmation state to PublisherCallbackChannel. Allows
the RabbitTemplate to handle confirms from multiple
channels.

Remove unnecessary synchronization.

Add tests for multiple threads - verifying the
functionality when more than one thread concurrently
uses the template.

NOTES: The CF cache size must be large enough
to cover the number of required channels - otherwise
the channel is closed before the confirms are received.

This could be alleviated by not closing channels if
P.C. are turned on.

Each PublisherCallbackConnection has two maps, the
first map (seqToListener) identifies which listener
gets the confirmation. The second map (pendingConfirms)
is a map of maps where the key is the listener and
the value is a map of pending confirms for that
listener. This map is given back to the listener
when registering.

In this way, the Rabbit template does not need to
hold a reference to every connection he has used.
Instead, when a connection is closed, the listener
is informed via a call to removePendingConfirmsReference().

The listener  uses the pending confirms maps to
determine if any confirms have expired.

Add test for two templates sharing the same
factory, ensuring the appropriate template
gets the confirmation.

AMQP-213 Polishing

AMQP-213 Add Support for Basic.Return

The rabbit template will be registered to receive return
messages if it is provided with a ReturnCallback. Either
mandatory or immediate must also be set for returns.

mandator and immediate are ignored if no callback is provided.

AMQP-213 Polishing

The RabbitTemplate can only support one each of
ConfirmCallback and ReturnCallback because there
is no way to correlate back to the caller.

In situations where more than one client is needed,
each needs its own RabbitTemplate.

The connection factory, and hence the connection and
channels, can be shared across templates.

AMQP-213 Add Namespace Support for Returns

Attributes added to template for mandatory, immediate,
return-callback.

Also confirm-callback for confirms.

AMQP-213 Return Reply Code Etc

Pass the reply code, reply text, exchange, and
routing key back to the client on returns.

AMQP-213 Polishing

Fix non-default exchange on convertAndSend.

AMQP-213 Polishing PR Review

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

@@ -43,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;
@@ -74,6 +78,9 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
 		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);
+
 	}
 
 }

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

@@ -13,6 +13,8 @@
 
 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;
@@ -20,6 +22,7 @@ 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;
@@ -50,6 +53,14 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser {
 
 	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;
@@ -87,9 +98,17 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser {
 		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 = getChildElement(element, parserContext);
+		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);
@@ -116,27 +135,8 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser {
 		}
 	}
 
-	private Element getChildElement(Element element,
-			ParserContext parserContext) {
-		Element childElement = null;
-		NodeList childNodes = element.getChildNodes();
-		for (int i = 0; i < childNodes.getLength(); i++) {
-			Node child = childNodes.item(i);
-			if (child.getNodeType() == Node.ELEMENT_NODE) {
-				String localName = parserContext.getDelegate().getLocalName(child);
-				if (LISTENER_ELEMENT.equals(localName)) {
-					childElement = (Element) child;
-				}
-			}
-		}
-		return childElement;
-	}
-
 	private BeanDefinition parseListener(Element childElement, Element element,
 			ParserContext parserContext) {
-		if (getChildElement(childElement, parserContext) != null) {
-			parserContext.getReaderContext().error("<reply-listener/> is not allowed any child elements.", element);
-		}
 		BeanDefinition replyContainer = RabbitNamespaceUtils.parseContainer(childElement, parserContext);
 		if (replyContainer != null) {
 			replyContainer.getPropertyValues().add(

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

@@ -13,6 +13,7 @@
 
 package org.springframework.amqp.rabbit.connection;
 
+import java.io.IOException;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -21,6 +22,8 @@ import java.util.LinkedList;
 import java.util.List;
 
 import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.rabbit.support.PublisherCallbackChannel;
+import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -57,6 +60,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory {
 
 	private ChannelCachingConnectionProxy connection;
 
+	private volatile boolean publisherConfirms;
+
 	/** Synchronization monitor for the shared Connection */
 	private final Object connectionMonitor = new Object();
 
@@ -118,6 +123,14 @@ public class CachingConnectionFactory extends AbstractConnectionFactory {
 		return this.channelCacheSize;
 	}
 
+	public boolean isPublisherConfirms() {
+		return publisherConfirms;
+	}
+
+	public void setPublisherConfirms(boolean publisherConfirms) {
+		this.publisherConfirms = publisherConfirms;
+	}
+
 	public void setConnectionListeners(List<? 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 {

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

@@ -14,7 +14,13 @@
 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;
@@ -33,14 +39,18 @@ 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;
@@ -82,7 +92,8 @@ import com.rabbitmq.client.GetResponse;
  * @author Gary Russell
  * @since 1.0
  */
-public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, MessageListener {
+public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, MessageListener,
+	PublisherCallbackChannel.Listener {
 
 	private static final String DEFAULT_EXCHANGE = ""; // alias for amq.direct default exchange
 
@@ -111,6 +122,18 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
 
 	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";
@@ -239,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);
 	}
@@ -248,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 {
@@ -274,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 {
@@ -425,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);
@@ -469,7 +570,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
 				if (logger.isDebugEnabled()) {
 					logger.debug("Sending message with tag " + messageTag);
 				}
-				doSend(channel, exchange, routingKey, message);
+				doSend(channel, exchange, routingKey, message, null);
 				Message reply = (replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout,
 						TimeUnit.MILLISECONDS);
 				RabbitTemplate.this.replyHolder.remove(messageTag);
@@ -483,6 +584,9 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
 		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);
@@ -507,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 + "]");
 		}
@@ -521,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.
@@ -562,6 +678,76 @@ 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);
@@ -637,4 +823,15 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
 			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);
+	}
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/CorrelationData.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.amqp.rabbit.support;
+
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+
+/**
+ * Base class for correlating publisher confirms to sent messages.
+ * Use the {@link RabbitTemplate} methods that include one of
+ * these as a parameter; when the publisher confirm is received,
+ * the CorrelationData is returned with the ack/nack.
+ * @author Gary Russell
+ * @since 1.0.1
+ *
+ */
+public class CorrelationData {
+
+	private String id;
+
+	public CorrelationData(String id) {
+		this.id = id;
+	}
+
+	public String getId() {
+		return id;
+	}
+
+	@Override
+	public String toString() {
+		return "CorrelationData [id=" + id + "]";
+	}
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.amqp.rabbit.support;
+
+/**
+ * Instances of this object track pending publisher confirms.
+ * The timestamp allows the pending confirmation to be
+ * expired. It also holds {@link CorrelationData} for
+ * the client to correlate a confirm with a sent message.
+ * @author Gary Russell
+ * @since 1.0.1
+ *
+ */
+public class PendingConfirm {
+
+	private final CorrelationData correlationData;
+
+	private final long timestamp;
+
+	/**
+	 * @param correlationId
+	 * @param timestamp
+	 */
+	public PendingConfirm(CorrelationData correlationData, long timestamp) {
+		this.correlationData = correlationData;
+		this.timestamp = timestamp;
+	}
+
+	public CorrelationData getCorrelationData() {
+		return correlationData;
+	}
+
+	public long getTimestamp() {
+		return timestamp;
+	}
+
+	@Override
+	public String toString() {
+		return "PendingConfirm [correlationData=" + correlationData + "]";
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.amqp.rabbit.support;
+
+import java.io.IOException;
+import java.util.SortedMap;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+
+/**
+ * Instances of this interface support a single listener being
+ * registered for publisher confirms with multiple channels,
+ * by adding context to the callbacks.
+ * @author Gary Russell
+ * @since 1.0.1
+ *
+ */
+public interface PublisherCallbackChannel extends Channel {
+
+	static String RETURN_CORRELATION = "spring_return_correlation";
+
+	/**
+	 * Adds a {@link Listener} and returns a reference to
+	 * the pending confirms map for that listener's pending
+	 * confirms, allowing the Listener to
+	 * assess unconfirmed sends at any point in time.
+	 * The client must <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();
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java

@@ -0,0 +1,559 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.amqp.rabbit.support;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.util.Assert;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AMQP.Basic.RecoverOk;
+import com.rabbitmq.client.AMQP.BasicProperties;
+import com.rabbitmq.client.AMQP.Channel.FlowOk;
+import com.rabbitmq.client.AMQP.Exchange.BindOk;
+import com.rabbitmq.client.AMQP.Exchange.DeclareOk;
+import com.rabbitmq.client.AMQP.Exchange.DeleteOk;
+import com.rabbitmq.client.AMQP.Exchange.UnbindOk;
+import com.rabbitmq.client.AMQP.Queue.PurgeOk;
+import com.rabbitmq.client.AMQP.Tx.CommitOk;
+import com.rabbitmq.client.AMQP.Tx.RollbackOk;
+import com.rabbitmq.client.AMQP.Tx.SelectOk;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Command;
+import com.rabbitmq.client.ConfirmListener;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.FlowListener;
+import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.client.Method;
+import com.rabbitmq.client.ReturnListener;
+import com.rabbitmq.client.ShutdownListener;
+import com.rabbitmq.client.ShutdownSignalException;
+
+/**
+ * Channel wrapper to allow a single listener able to handle
+ * confirms from multiple channels.
+ *
+ * @author Gary Russell
+ * @since 1.0.1
+ *
+ */
+public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final Channel delegate;
+
+	private final Map<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();
+	}
+
+}

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

@@ -746,6 +746,50 @@
 					</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>
 
@@ -839,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>
 

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

@@ -13,7 +13,9 @@
 
 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;
 
@@ -47,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

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java

@@ -111,6 +111,7 @@ public class RabbitTemplateHeaderTests {
 				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);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java

@@ -0,0 +1,351 @@
+/*
+ * Copyright 2010-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.core;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
+import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
+import org.springframework.amqp.rabbit.support.CorrelationData;
+import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
+import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
+import org.springframework.amqp.support.converter.SimpleMessageConverter;
+import org.springframework.beans.DirectFieldAccessor;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+public class RabbitTemplatePublisherCallbacksIntegrationTests {
+
+	private static final String ROUTE = "test.queue";
+
+	private CachingConnectionFactory connectionFactory;
+
+	private RabbitTemplate template;
+
+	@Before
+	public void create() {
+		connectionFactory = new CachingConnectionFactory();
+		// When using publisher confirms, the cache size needs to be large enough
+		// otherwise channels can be closed before confirms are received.
+		connectionFactory.setChannelCacheSize(10);
+		connectionFactory.setPort(BrokerTestUtils.getPort());
+		connectionFactory.setPublisherConfirms(true);
+		template = new RabbitTemplate(connectionFactory);
+	}
+
+	@Rule
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE);
+
+	@Test
+	public void testPublisherConfirmReceived() throws Exception {
+		final CountDownLatch latch = new CountDownLatch(1);
+		template.setConfirmCallback(new ConfirmCallback() {
+
+			public void confirm(CorrelationData correlationData, boolean ack) {
+				latch.countDown();
+			}
+		});
+		template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
+		assertTrue(latch.await(1000, TimeUnit.MILLISECONDS));
+		assertNull(template.getUnconfirmed(0));
+	}
+
+	@Test
+	public void testPublisherConfirmReceivedConcurrentThreads() throws Exception {
+		final CountDownLatch latch = new CountDownLatch(2);
+		template.setConfirmCallback(new ConfirmCallback() {
+
+			public void confirm(CorrelationData correlationData, boolean ack) {
+				latch.countDown();
+			}
+		});
+
+		// Hold up the first thread so we get two channels
+		final CountDownLatch threadLatch = new CountDownLatch(1);
+		//Thread 1
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
+
+			public void run() {
+				template.execute(new ChannelCallback<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);
+	}
+}

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

@@ -23,4 +23,14 @@
 		<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>