diff --git a/build.gradle b/build.gradle
index e8e74244f1..13442b5748 100644
--- a/build.gradle
+++ b/build.gradle
@@ -47,7 +47,7 @@ subprojects { subproject ->
 		mockitoVersion = '1.9.0'
 		// When changing Spring Versions - don't forget to update bundlor ranges
 		springVersion = '3.1.1.RELEASE'
-		springAmqpVersion = '1.0.0.RELEASE'
+		springAmqpVersion = '1.1.0.RELEASE'
 		springDataMongoVersion = '1.1.0.M1'
 		springDataRedisVersion = '1.0.0.RELEASE'
 		springGemfireVersion = '1.1.0.RELEASE'
diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java
index 0d1b3460a2..fefd1a1912 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java
@@ -72,4 +72,18 @@ public abstract class AmqpHeaders {
 
 	public static final String USER_ID = PREFIX + "userId";
 
+	public static final String SPRING_REPLY_CORRELATION = PREFIX + "springReplyCorrelation";
+
+	public static final String SPRING_REPLY_TO_STACK = PREFIX + "springReplyToStack";
+
+	public static final String PUBLISH_CONFIRM = PREFIX + "publishConfirm";
+
+	public static final String RETURN_REPLY_CODE = PREFIX + "returnReplyCode";
+
+	public static final String RETURN_REPLY_TEXT = PREFIX + "returnReplyText";
+
+	public static final String RETURN_EXCHANGE = PREFIX + "returnExchange";
+
+	public static final String RETURN_ROUTING_KEY = PREFIX + "returnRoutingKey";
+
 }
diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
index 3a104220aa..63ccd94067 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import org.w3c.dom.Element;
  * 
  * @author Mark Fisher
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  * @since 2.1
  */
 public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@@ -47,9 +48,14 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name-expression");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
-	
+
 		IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
-		
+
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-correlation-expression");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
+
 		return builder.getBeanDefinition();
 	}
 
diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
index f89bd2983a..0a712647eb 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
@@ -51,7 +51,8 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
-		
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
+
 		IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
 		
 		return builder;
diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
index e3972c7cf2..bb912ea81c 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
@@ -20,12 +20,16 @@ import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
+import org.springframework.amqp.rabbit.support.CorrelationData;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.expression.Expression;
 import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.SpelParserConfiguration;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.amqp.AmqpHeaders;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -41,7 +45,8 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @since 2.1
  */
-public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
+public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
+	implements RabbitTemplate.ConfirmCallback, ReturnCallback {
 
 	private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
 
@@ -64,11 +69,23 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
 
 	private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
 
+	private volatile String confirmCorrelationExpression;
+
+	private volatile ExpressionEvaluatingMessageProcessor<Object> correlationDataGenerator;
+
+	private volatile MessageChannel confirmAckChannel;
+
+	private volatile MessageChannel confirmNackChannel;
+
+	private volatile MessageChannel returnChannel;
+
 	@Override
 	protected void onInit() {
 		super.onInit();
 		Assert.state(exchangeNameExpression == null || "".equals(exchangeName),
 				"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
+		Assert.state(this.confirmCorrelationExpression != null ? !this.expectReply : true,
+				"Confirm correlation expression does not apply to a gateway");
 		if (exchangeNameExpression != null) {
 			Expression expression = expressionParser.parseExpression(this.exchangeNameExpression);
 			this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
@@ -79,6 +96,16 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
 			Expression expression = expressionParser.parseExpression(this.routingKeyExpression);
 			this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
 		}
+		if (this.confirmCorrelationExpression != null) {
+			Expression expression = expressionParser.parseExpression(this.confirmCorrelationExpression);
+			this.correlationDataGenerator = new ExpressionEvaluatingMessageProcessor<Object>(expression, Object.class);
+			Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for publisher confirms");
+			((RabbitTemplate) this.amqpTemplate).setConfirmCallback(this);
+		}
+		if (this.returnChannel != null) {
+			Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for publisher returns");
+			((RabbitTemplate) this.amqpTemplate).setReturnCallback(this);
+		}
 	}
 
 	public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) {
@@ -110,6 +137,22 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
 		this.expectReply = expectReply;
 	}
 
+	public void setConfirmCorrelationExpression(String confirmCorrelationExpression) {
+		this.confirmCorrelationExpression = confirmCorrelationExpression;
+	}
+
+	public void setConfirmAckChannel(MessageChannel ackChannel) {
+		this.confirmAckChannel = ackChannel;
+	}
+
+	public void setConfirmNackChannel(MessageChannel nackChannel) {
+		this.confirmNackChannel = nackChannel;
+	}
+
+	public void setReturnChannel(MessageChannel returnChannel) {
+		this.returnChannel = returnChannel;
+	}
+
 	@Override
 	public String getComponentType() {
 		return expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter";
@@ -119,6 +162,19 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		String exchangeName = this.exchangeName;
 		String routingKey = this.routingKey;
+		CorrelationData correlationData = null;
+		if (this.correlationDataGenerator != null) {
+			Object userCorrelationData = this.correlationDataGenerator
+					.processMessage(requestMessage);
+			if (userCorrelationData != null) {
+				if (userCorrelationData instanceof CorrelationData) {
+					correlationData = (CorrelationData) userCorrelationData;
+				} else {
+					correlationData = new CorrelationDataWrapper(requestMessage
+							.getHeaders().getId().toString(), userCorrelationData);
+				}
+			}
+		}
 		if (this.exchangeNameGenerator != null) {
 			exchangeName = this.exchangeNameGenerator.processMessage(requestMessage);
 		}
@@ -129,20 +185,34 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
 			return this.sendAndReceive(exchangeName, routingKey, requestMessage);
 		}
 		else {
-			this.send(exchangeName, routingKey, requestMessage);
+			this.send(exchangeName, routingKey, requestMessage, correlationData);
 			return null;
 		}
 	}
 
-	private void send(String exchangeName, String routingKey, final Message<?> requestMessage) {
-		this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
-				new MessagePostProcessor() {
-					public org.springframework.amqp.core.Message postProcessMessage(
-							org.springframework.amqp.core.Message message) throws AmqpException {
-						headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
-						return message;
-					}
-				});
+	private void send(String exchangeName, String routingKey,
+			final Message<?> requestMessage, CorrelationData correlationData) {
+		if (this.amqpTemplate instanceof RabbitTemplate) {
+			((RabbitTemplate) this.amqpTemplate).convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
+					new MessagePostProcessor() {
+						public org.springframework.amqp.core.Message postProcessMessage(
+								org.springframework.amqp.core.Message message) throws AmqpException {
+							headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
+							return message;
+						}
+					},
+					correlationData);
+		}
+		else {
+			this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
+					new MessagePostProcessor() {
+						public org.springframework.amqp.core.Message postProcessMessage(
+								org.springframework.amqp.core.Message message) throws AmqpException {
+							headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
+							return message;
+						}
+					});
+		}
 	}
 
 	private Message<?> sendAndReceive(String exchangeName, String routingKey, Message<?> requestMessage) {
@@ -165,4 +235,58 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
 		return builder.build();
 	}
 
+	public void confirm(CorrelationData correlationData, boolean ack) {
+		Object userCorrelationData = correlationData;
+		if (correlationData instanceof CorrelationDataWrapper) {
+			userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData();
+		}
+		Message<Object> confirmMessage = MessageBuilder.withPayload(userCorrelationData)
+				.setHeader(AmqpHeaders.PUBLISH_CONFIRM, ack)
+				.build();
+		if (ack && this.confirmAckChannel != null) {
+			this.confirmAckChannel.send(confirmMessage);
+		}
+		else if (!ack && this.confirmNackChannel != null) {
+			this.confirmNackChannel.send(confirmMessage);
+		}
+		else {
+			if (logger.isInfoEnabled()) {
+				logger.info("Nowhere to send publisher confirm "
+						+ (ack ? "ack" : "nack") + " for "
+						+ userCorrelationData);
+			}
+		}
+	}
+
+	private static class CorrelationDataWrapper extends CorrelationData {
+
+		private final Object userData;
+
+		public CorrelationDataWrapper(String id, Object userData) {
+			super(id);
+			this.userData = userData;
+		}
+
+		public Object getUserData() {
+			return userData;
+		}
+
+	}
+
+	public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
+			String exchange, String routingKey) {
+		// safe to cast; we asserted we have a RabbitTemplate in onInit()
+		MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
+		Object returnedObject = converter.fromMessage(message);
+		MessageBuilder<?> builder = (returnedObject instanceof Message)
+				? MessageBuilder.fromMessage((Message<?>) returnedObject)
+				: MessageBuilder.withPayload(returnedObject);
+		Map<String, ?> headers = this.headerMapper.toHeadersFromReply(message.getMessageProperties());
+		builder.copyHeadersIfAbsent(headers)
+			.setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)
+			.setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText)
+			.setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange)
+			.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey);
+		this.returnChannel.send(builder.build());
+	}
 }
diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
index fd61b37d5e..4c74d11459 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
@@ -24,6 +24,7 @@ import java.util.Map;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.amqp.AmqpHeaders;
 import org.springframework.integration.mapping.AbstractHeaderMapper;
@@ -70,6 +71,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		STANDARD_HEADER_NAMES.add(AmqpHeaders.TIMESTAMP);
 		STANDARD_HEADER_NAMES.add(AmqpHeaders.TYPE);
 		STANDARD_HEADER_NAMES.add(AmqpHeaders.USER_ID);
+		STANDARD_HEADER_NAMES.add(AmqpHeaders.SPRING_REPLY_CORRELATION);
+		STANDARD_HEADER_NAMES.add(AmqpHeaders.SPRING_REPLY_TO_STACK);
 	}
 
 	/**
@@ -155,6 +158,18 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 			if (StringUtils.hasText(userId)) {
 				headers.put(AmqpHeaders.USER_ID, userId);
 			}
+			Object replyCorrelation = amqpMessageProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER);
+			if (replyCorrelation instanceof String) {
+				if (StringUtils.hasText((String) replyCorrelation)) {
+					headers.put(AmqpHeaders.SPRING_REPLY_CORRELATION, replyCorrelation);
+				}
+			}
+			Object replyToStack = amqpMessageProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER);
+			if (replyToStack instanceof String) {
+				if (StringUtils.hasText((String) replyToStack)) {
+					headers.put(AmqpHeaders.SPRING_REPLY_TO_STACK, replyToStack);
+				}
+			}
 		}
 		catch (Exception e) {
 			if (logger.isWarnEnabled()) {
@@ -169,7 +184,10 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 	 */
 	@Override
 	protected Map<String, Object> extractUserDefinedHeaders(MessageProperties amqpMessageProperties) {
-		return amqpMessageProperties.getHeaders();
+		Map<String, Object> headers = amqpMessageProperties.getHeaders();
+		headers.remove(RabbitTemplate.STACKED_CORRELATION_HEADER);
+		headers.remove(RabbitTemplate.STACKED_REPLY_TO_HEADER);
+		return headers;
 	}
 
 	/**
@@ -254,6 +272,14 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		if (StringUtils.hasText(userId)) {
 			amqpMessageProperties.setUserId(userId);
 		}
+		String replyCorrelation = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class);
+		if (StringUtils.hasLength(replyCorrelation)) {
+			amqpMessageProperties.setHeader("spring_reply_correlation", replyCorrelation);
+		}
+		String replyToStack = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_TO_STACK, String.class);
+		if (StringUtils.hasLength(replyToStack)) {
+			amqpMessageProperties.setHeader("spring_reply_to", replyToStack);
+		}
 	}
 
 	@Override
diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.2.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.2.xsd
index d93534ae98..44ba3adeec 100644
--- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.2.xsd
+++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.2.xsd
@@ -18,91 +18,75 @@
 			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
-			<xsd:attribute name="id" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	Unique ID for this adapter.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="channel" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
-	If this attribute is not provided, the ID will be used to create a new DirectChannel, and then instead of using that
-	ID as the bean name of the EventDrivenConsumer instance that hosts the MessageHandler responsible for publishing the
-	AMQP Messages, that EventDrivenConsumer's bean name will be the ID plus the added suffix: ".adapter"
-					</xsd:documentation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type
-								type="org.springframework.integration.MessageChannel" />
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="exchange-name" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The fixed name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="exchange-name-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="routing-key" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The fixed routing-key to use when sending Messages. By default, this will be an empty String.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="routing-key-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The routing-key to use when sending Messages evaluated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="amqp-template" type="xsd:string">
-				<xsd:annotation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate" />
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="order" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover. 
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="header-mapper" type="xsd:string">
-			<xsd:annotation>
-				<xsd:appinfo>
-					<tool:annotation kind="ref">
-						<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
-					</tool:annotation>
-				</xsd:appinfo>
-			</xsd:annotation>
-		</xsd:attribute>
-		<xsd:attribute name="mapped-request-headers" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation><![CDATA[
-Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
-This can only be provided if the 'header-mapper' reference is not being set directly. The values in
-this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
-						]]></xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
+			<xsd:complexContent>
+				<xsd:extension base="outboundType">
+					<xsd:attribute name="id" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+			Unique ID for this adapter.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="channel" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+			Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
+			If this attribute is not provided, the ID will be used to create a new DirectChannel, and then instead of using that
+			ID as the bean name of the EventDrivenConsumer instance that hosts the MessageHandler responsible for publishing the
+			AMQP Messages, that EventDrivenConsumer's bean name will be the ID plus the added suffix: ".adapter"
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.integration.MessageChannel" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="confirm-correlation-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+			Expression for correlating publisher confirms to sent messages. The Rabbit API only correlates confirms
+			to a channel; this is used to further correlate a confirm to a message. An example is
+			"headers['amqp_confirmCorrelationData']", assuming that header contains the data.
+			Messages that do not have correlation data do not generate publisher confirm messages.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="confirm-ack-channel" type="xsd:string" default="nullChannel">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+			Channel to which positive publisher confirms will be sent; the payload will be the correlation data from
+			the sent message. The message will also contain a header 'amqp_publishConfirm' with a value true.
+			Requires a connection factory that is configured to request publisher confirms. Default
+			is nullChannel in case an adapter doesn't want to use publisher confirms, but is using a
+			connection factory that is configured to request them.
+							]]></xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="confirm-nack-channel" type="xsd:string" default="nullChannel">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+			Channel to which negative publisher confirms will be sent; the payload will be the correlation data from
+			the sent message. The message will also contain a header 'amqp_publishConfirm' with a value false.
+			Requires a connection factory that is configured to request publisher confirms. Default
+			is nullChannel in case an adapter doesn't want to use publisher confirms, but is using a
+			connection factory that is configured to request them.
+							]]></xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>
 
@@ -142,112 +126,54 @@ this list can also be simple patterns to be matched against the header names (e.
             </xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
-			<xsd:attribute name="id" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	Unique ID for this gateway.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="request-channel" use="required" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
-					</xsd:documentation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type
-								type="org.springframework.integration.MessageChannel"/>
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="reply-channel" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	Message Channel to which replies should be sent after being received from an AQMP Queue and converted.
-					</xsd:documentation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type
-								type="org.springframework.integration.MessageChannel"/>
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="exchange-name" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="exchange-name-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="routing-key" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The routing-key to use when sending Messages. By default, this will be an empty String.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="routing-key-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The routing-key to use when sending Messages evealuated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="amqp-template" type="xsd:string">
-				<xsd:annotation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate"/>
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="order" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-	The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover. 
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="header-mapper" type="xsd:string">
-				<xsd:annotation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="mapped-request-headers" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
-This can only be provided if the 'header-mapper' reference is not being set directly. The values in
-this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
-						]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="mapped-reply-headers" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[							
-    Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
-	All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property 
-	which itself is a Map.
-	This can only be provided if the 'header-mapper' reference is not being set directly. The values in
-	this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
-						]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
+			<xsd:complexContent>
+				<xsd:extension base="outboundType">
+					<xsd:attribute name="id" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+			Unique ID for this gateway.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="request-channel" use="required" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+			Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange.
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.integration.MessageChannel"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="reply-channel" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+			Message Channel to which replies should be sent after being received from an AQMP Queue and converted.
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.integration.MessageChannel"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="mapped-reply-headers" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+			Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
+			All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
+			which itself is a Map.
+			This can only be provided if the 'header-mapper' reference is not being set directly. The values in
+			this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
+								]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>
 
@@ -287,9 +213,9 @@ this list can also be simple patterns to be matched against the header names (e.
 					</xsd:attribute>
 					<xsd:attribute name="mapped-reply-headers" type="xsd:string">
 						<xsd:annotation>
-							<xsd:documentation><![CDATA[							
+							<xsd:documentation><![CDATA[
 	Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
-	All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property 
+	All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
 	which itself is a Map.
 	This can only be provided if the 'header-mapper' reference is not being set directly. The values in
 	this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
@@ -365,7 +291,7 @@ this list can also be simple patterns to be matched against the header names (e.
 	Reference to a FanoutExchange instance to which this channel should send Messages. If not provided,
 	a FanoutExchange will be declared with this channel's name prefixed by "si.fanout.".
 	A Queue will be declared automatically and bound to that exchange to handle the consumer role
-	of this channel. 
+	of this channel.
 							</xsd:documentation>
 							<xsd:appinfo>
 								<tool:annotation kind="ref">
@@ -412,6 +338,88 @@ this list can also be simple patterns to be matched against the header names (e.
 		<xsd:attributeGroup ref="containerAndTemplateAttributes"/>
 	</xsd:complexType>
 
+	<xsd:complexType name="outboundType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Base type for the 'outbound-channel-adapter' and 'outbound-gateway' elements.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="exchange-name" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+The fixed name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="exchange-name-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="routing-key" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+The fixed routing-key to use when sending Messages. By default, this will be an empty String.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="routing-key-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+The routing-key to use when sending Messages evaluated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="amqp-template" type="xsd:string">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.amqp.core.AmqpTemplate" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="order" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+The order for this consumer when multiple consumers are registered thereby enabling load-balancing and/or failover.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="header-mapper" type="xsd:string">
+			<xsd:annotation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="mapped-request-headers" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
+This can only be provided if the 'header-mapper' reference is not being set directly. The values in
+this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
+						]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="return-channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Channel to which returned messages will be sent.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
 	<xsd:complexType name="inboundType">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -527,7 +535,7 @@ this list can also be simple patterns to be matched against the header names (e.
 		<xsd:attribute name="connection-factory" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
-	Reference to the Rabbit ConnectionFactory to be used by this component.	
+	Reference to the Rabbit ConnectionFactory to be used by this component.
 				</xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
@@ -583,7 +591,7 @@ this list can also be simple patterns to be matched against the header names (e.
 	<xsd:attributeGroup name="containerOnlyAttributes">
 		<xsd:annotation>
 			<xsd:documentation>
-	Attributes for a SimpleMesssageListenerContainer's properties other than queues, queueNames, messageListener, and 
+	Attributes for a SimpleMesssageListenerContainer's properties other than queues, queueNames, messageListener, and
 	autoStartup which may or may not be exposed for configuration depending on what type of component uses this attribute group.
 	This group also does not include any of the properties that are shared with RabbitTemplate, such as channelTransacted,
 	connectionFactory, and messsagePropertiesConverter.
@@ -692,7 +700,7 @@ this list can also be simple patterns to be matched against the header names (e.
 					</xsd:documentation>
 				</xsd:appinfo>
 			</xsd:annotation>
-		</xsd:attribute>	
+		</xsd:attribute>
 		<xsd:attribute name="task-executor" type="xsd:string">
 			<xsd:annotation>
 				<xsd:appinfo>
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
index eec2111aec..c4bcf45cc6 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
@@ -18,6 +18,7 @@ package org.springframework.integration.amqp;
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.concurrent.TimeoutException;
 
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.rabbit.connection.Connection;
@@ -25,6 +26,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionListener;
 
 import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Command;
 import com.rabbitmq.client.ConfirmListener;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.FlowListener;
@@ -47,6 +49,7 @@ import com.rabbitmq.client.AMQP.Tx.RollbackOk;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.1
  */
 public class StubRabbitConnectionFactory implements ConnectionFactory {
@@ -132,24 +135,29 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
 		public void abort(int closeCode, String closeMessage) throws IOException {
 		}
 
+		@SuppressWarnings("unused")
 		public ReturnListener getReturnListener() {
 			return null;
 		}
 
-		public void setReturnListener(ReturnListener listener) {
+		public void addReturnListener(ReturnListener listener) {
 		}
 
+		@SuppressWarnings("unused")
 		public FlowListener getFlowListener() {
 			return null;
 		}
 
+		@SuppressWarnings("unused")
 		public void setFlowListener(FlowListener listener) {
 		}
 
+		@SuppressWarnings("unused")
 		public ConfirmListener getConfirmListener() {
 			return null;
 		}
 
+		@SuppressWarnings("unused")
 		public void setConfirmListener(ConfirmListener listener) {
 		}
 
@@ -354,9 +362,53 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
 		public void asyncRpc(Method method) throws IOException {
 		}
 
-		public Method rpc(Method method) throws IOException {
+		public Command rpc(Method method) throws IOException {
 			return null;
 		}
+
+		public boolean removeReturnListener(ReturnListener listener) {
+			return false;
+		}
+
+		public void clearReturnListeners() {
+		}
+
+		public void addFlowListener(FlowListener listener) {
+		}
+
+		public boolean removeFlowListener(FlowListener listener) {
+			return false;
+		}
+
+		public void clearFlowListeners() {
+		}
+
+		public void addConfirmListener(ConfirmListener listener) {
+		}
+
+		public boolean removeConfirmListener(ConfirmListener listener) {
+			return false;
+		}
+
+		public void clearConfirmListeners() {
+		}
+
+		public boolean waitForConfirms() throws InterruptedException {
+			return false;
+		}
+
+		public void waitForConfirmsOrDie() throws IOException,
+				InterruptedException {
+		}
+
+		public boolean waitForConfirms(long timeout)
+				throws InterruptedException, TimeoutException {
+			return false;
+		}
+
+		public void waitForConfirmsOrDie(long timeout) throws IOException,
+				InterruptedException, TimeoutException {
+		}
 	}
 
 }
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java
index faf9eb11a2..807f09ede5 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java
@@ -28,6 +28,7 @@ import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.support.CorrelationData;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -105,7 +106,8 @@ public class AmqpInboundGatewayParserTests {
 		          assertEquals("bar", properties.getHeaders().get("bar"));
 		          return null;
 		      }})
-		 .when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class));
+		 .when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
+				 Mockito.any(Message.class), Mockito.any(CorrelationData.class));
 		ReflectionUtils.setField(amqpTemplateField, gateway, amqpTemplate);
 		
 		AbstractMessageListenerContainer mlc = 
@@ -123,7 +125,8 @@ public class AmqpInboundGatewayParserTests {
 		Message amqpMessage = new Message("hello".getBytes(), amqpProperties);
 		listener.onMessage(amqpMessage);
 		
-		Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class));
+		Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class),
+				Mockito.any(Message.class), Mockito.any(CorrelationData.class));
 	}
 
 	private static class TestConverter extends SimpleMessageConverter {}
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
index 0d3cb3947d..056f6c7bdd 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
@@ -16,16 +16,38 @@
 	<bean id="connectionFactory" class="org.mockito.Mockito" factory-method="mock">
 		<constructor-arg value="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
 	</bean>
-	
-	<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel" 
+
+	<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel"
 								   exchange-name="outboundchanneladapter.test.1"
 								   mapped-request-headers="foo*"/>
 								   
-    <int:channel id="requestChannel"/>
+	<int:channel id="requestChannel"/>
 
 	<int:chain id="chainWithRabbitOutbound" input-channel="amqpOutboundChannelAdapterWithinChain">
 		<amqp:outbound-channel-adapter exchange-name="outboundchanneladapter.test.1"/>
 	</int:chain>
 
+	<amqp:outbound-channel-adapter id="withPublisherConfirms" channel="pcRequestChannel"
+								   exchange-name="outboundchanneladapter.test.1"
+								   mapped-request-headers="foo*"
+								   confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
+								   confirm-ack-channel="ackChannel"/>
+
+	<int:channel id="pcRequestChannel"/>
+
+	<int:channel id="ackChannel">
+		<int:queue/>
+	</int:channel>
+
+	<amqp:outbound-channel-adapter id="withReturns" channel="returnRequestChannel"
+								   exchange-name="outboundchanneladapter.test.1"
+								   mapped-request-headers="foo*"
+								   return-channel="returnChannel"/>
+
+	<int:channel id="returnRequestChannel"/>
+
+	<int:channel id="returnChannel">
+		<int:queue/>
+	</int:channel>
 
 </beans>
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
index e273dbb21f..c9673f071d 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
@@ -16,8 +16,17 @@
 
 package org.springframework.integration.amqp.config;
 
+import static junit.framework.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 import java.lang.reflect.Field;
 import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -25,14 +34,22 @@ import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.Connection;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.CorrelationData;
+import org.springframework.amqp.rabbit.support.PublisherCallbackChannel;
+import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
+import org.springframework.integration.amqp.AmqpHeaders;
 import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.handler.MessageHandlerChain;
 import org.springframework.integration.support.MessageBuilder;
@@ -41,9 +58,8 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.ReflectionUtils;
 
-import static org.junit.Assert.assertNull;
-
-import static junit.framework.Assert.assertEquals;
+import com.rabbitmq.client.AMQP.BasicProperties;
+import com.rabbitmq.client.Channel;
 
 /**
  * @author Mark Fisher
@@ -70,36 +86,70 @@ public class AmqpOutboundChannelAdapterParserTests {
 		assertEquals("amqp:outbound-channel-adapter", ((AmqpOutboundEndpoint) handler).getComponentType());
 	}
 	
-	@SuppressWarnings("rawtypes")
 	@Test
 	public void withHeaderMapperCustomHeaders() {
-		Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomHeaders");
+		Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomHeaders");
 		
-		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
+		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
 		
 		Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
 		amqpTemplateField.setAccessible(true);
 		RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
 		amqpTemplate = Mockito.spy(amqpTemplate);
 
-		Mockito.doAnswer(new Answer() {
-		      public Object answer(InvocationOnMock invocation) {
-		          Object[] args = invocation.getArguments();
-		          org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2];
-		          MessageProperties properties = amqpReplyMessage.getMessageProperties();
-		          assertEquals("foo", properties.getHeaders().get("foo"));
-		          assertEquals("foobar", properties.getHeaders().get("foobar"));
-		          assertNull(properties.getHeaders().get("bar"));
-		          return null;
-		      }})
-		 .when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
+		Mockito.doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) {
+				Object[] args = invocation.getArguments();
+				org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2];
+				MessageProperties properties = amqpReplyMessage.getMessageProperties();
+				assertEquals("foo", properties.getHeaders().get("foo"));
+				assertEquals("foobar", properties.getHeaders().get("foobar"));
+				assertNull(properties.getHeaders().get("bar"));
+				return null;
+			}})
+		 .when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
+				 Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class));
 		ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
 		
 		
 		MessageChannel requestChannel = context.getBean("requestChannel", MessageChannel.class);
 		Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").setHeader("bar", "bar").setHeader("foobar", "foobar").build();
 		requestChannel.send(message);
-		Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
+		Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class),
+				Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class));
+	}
+
+	@Test
+	public void parseWithPublisherConfirms() {
+		Object eventDrivenConsumer = context.getBean("withPublisherConfirms");
+		AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
+		NullChannel nullChannel = context.getBean(NullChannel.class);
+		MessageChannel ackChannel = context.getBean("ackChannel", MessageChannel.class);
+		assertSame(ackChannel, TestUtils.getPropertyValue(endpoint, "confirmAckChannel"));
+		assertSame(nullChannel, TestUtils.getPropertyValue(endpoint, "confirmNackChannel"));
+	}
+
+	@Test
+	public void withPublisherConfirms() throws Exception {
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+		Connection mockConnection = mock(Connection.class);
+		Channel mockChannel = mock(Channel.class);
+
+		when(connectionFactory.createConnection()).thenReturn(mockConnection);
+		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
+		when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
+
+		MessageChannel requestChannel = context.getBean("pcRequestChannel", MessageChannel.class);
+		Message<?> message = MessageBuilder.withPayload("hello")
+				.setHeader("amqp_confirmCorrelationData", "foo")
+				.build();
+		requestChannel.send(message);
+		PollableChannel ackChannel = context.getBean("ackChannel", PollableChannel.class);
+		publisherCallbackChannel.handleAck(0, false);
+		Message<?> ack = ackChannel.receive(1000);
+		assertNotNull(ack);
+		assertEquals("foo", ack.getPayload());
+		assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
 	}
 
 	@SuppressWarnings("rawtypes")
@@ -123,14 +173,45 @@ public class AmqpOutboundChannelAdapterParserTests {
 				assertEquals("hello", new String(amqpReplyMessage.getBody()));
 				return null;
 			}})
-				.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
+				.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class),
+						Mockito.any(CorrelationData.class));
 		ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
 
 
 		MessageChannel requestChannel = context.getBean("amqpOutboundChannelAdapterWithinChain", MessageChannel.class);
 		Message<?> message = MessageBuilder.withPayload("hello").build();
 		requestChannel.send(message);
-		Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
+		Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class),
+				Mockito.any(CorrelationData.class));
 	}
 
+	@Test
+	public void withReturns() throws Exception {
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+		Connection mockConnection = mock(Connection.class);
+		Channel mockChannel = mock(Channel.class);
+
+		when(connectionFactory.createConnection()).thenReturn(mockConnection);
+		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
+		when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
+
+		MessageChannel requestChannel = context.getBean("returnRequestChannel", MessageChannel.class);
+		Message<?> message = MessageBuilder.withPayload("hello").build();
+		requestChannel.send(message);
+		PollableChannel returnChannel = context.getBean("returnChannel", PollableChannel.class);
+		RabbitTemplate template = context.getBean(RabbitTemplate.class);
+		Map<String, Object> headers = new HashMap<String, Object>();
+		headers.put(PublisherCallbackChannel.RETURN_CORRELATION, template.getUUID());
+		BasicProperties properties = mock(BasicProperties.class);
+		when(properties.getHeaders()).thenReturn(headers);
+		when(properties.getContentType()).thenReturn("text/plain");
+		publisherCallbackChannel.handleReturn(123, "reply text", "anExchange", "bar", properties, "hello".getBytes());
+		Message<?> returned = returnChannel.receive(1000);
+		assertNotNull(returned);
+		assertEquals(123, returned.getHeaders().get(AmqpHeaders.RETURN_REPLY_CODE));
+		assertEquals("reply text", returned.getHeaders().get(AmqpHeaders.RETURN_REPLY_TEXT));
+		assertEquals("anExchange", returned.getHeaders().get(AmqpHeaders.RETURN_EXCHANGE));
+		assertEquals("bar", returned.getHeaders().get(AmqpHeaders.RETURN_ROUTING_KEY));
+		assertEquals("hello", returned.getPayload());
+	}
 }
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml
index 505457a150..df3572f9a6 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml
@@ -14,7 +14,8 @@
 		exchange-name="si.test.exchange"
 		routing-key="si.test.binding"
 		amqp-template="amqpTemplate" 
-		order="5"/>
+		order="5"
+		return-channel="returnChannel"/>
 		
 	<rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>
 	
@@ -54,4 +55,8 @@
 		mapped-request-headers=""
 		mapped-reply-headers=""/>
 
+	<int:channel id="returnChannel">
+		<int:queue/>
+	</int:channel>
+
 </beans>
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
index 7c9ab92417..d35901fa1d 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
@@ -38,6 +38,7 @@ import static junit.framework.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;
 
 /**
@@ -57,6 +58,8 @@ public class AmqpOutboundGatewayParserTests {
 		assertTrue(context.containsBean("rabbitGateway"));
 		assertEquals(context.getBean("fromRabbit"), TestUtils.getPropertyValue(gateway, "outputChannel"));
 		assertEquals("amqp:outbound-gateway", gateway.getComponentType());
+		MessageChannel returnChannel = context.getBean("returnChannel", MessageChannel.class);
+		assertSame(returnChannel, TestUtils.getPropertyValue(gateway, "returnChannel"));
 	}
 	
 	@SuppressWarnings("rawtypes")
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
index 8e15f680d7..c83eb36eda 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,6 +26,7 @@ import org.junit.Test;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.support.converter.JsonMessageConverter;
 import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.amqp.AmqpHeaders;
@@ -35,6 +36,7 @@ import static org.junit.Assert.fail;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.1
  */
 public class DefaultAmqpHeaderMapperTests {
@@ -62,6 +64,8 @@ public class DefaultAmqpHeaderMapperTests {
 		headerMap.put(AmqpHeaders.TIMESTAMP, testTimestamp);
 		headerMap.put(AmqpHeaders.TYPE, "test.type");
 		headerMap.put(AmqpHeaders.USER_ID, "test.userId");
+		headerMap.put(AmqpHeaders.SPRING_REPLY_CORRELATION, "test.correlation");
+		headerMap.put(AmqpHeaders.SPRING_REPLY_TO_STACK, "test.replyTo2");
 		MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
 		MessageProperties amqpProperties = new MessageProperties();
 		headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
@@ -88,6 +92,8 @@ public class DefaultAmqpHeaderMapperTests {
 		assertEquals(testTimestamp, amqpProperties.getTimestamp());
 		assertEquals("test.type", amqpProperties.getType());
 		assertEquals("test.userId", amqpProperties.getUserId());
+		assertEquals("test.correlation", amqpProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER));
+		assertEquals("test.replyTo2", amqpProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER));
 	}
 
 	@Test
@@ -115,6 +121,8 @@ public class DefaultAmqpHeaderMapperTests {
 		amqpProperties.setTimestamp(testTimestamp);
 		amqpProperties.setType("test.type");
 		amqpProperties.setUserId("test.userId");
+		amqpProperties.setHeader(RabbitTemplate.STACKED_CORRELATION_HEADER, "test.correlation");
+		amqpProperties.setHeader(RabbitTemplate.STACKED_REPLY_TO_HEADER, "test.replyTo2");
 		Map<String, Object> headerMap = headerMapper.toHeadersFromReply(amqpProperties);
 		assertEquals("test.appId", headerMap.get(AmqpHeaders.APP_ID));
 		assertEquals("test.clusterId", headerMap.get(AmqpHeaders.CLUSTER_ID));
@@ -133,6 +141,8 @@ public class DefaultAmqpHeaderMapperTests {
 		assertEquals(testTimestamp, headerMap.get(AmqpHeaders.TIMESTAMP));
 		assertEquals("test.type", headerMap.get(AmqpHeaders.TYPE));
 		assertEquals("test.userId", headerMap.get(AmqpHeaders.USER_ID));
+		assertEquals("test.correlation", headerMap.get(AmqpHeaders.SPRING_REPLY_CORRELATION));
+		assertEquals("test.replyTo2", headerMap.get(AmqpHeaders.SPRING_REPLY_TO_STACK));
 	}
 
 	@Test
@@ -171,4 +181,5 @@ public class DefaultAmqpHeaderMapperTests {
 		Object result = converter.fromMessage(new Message("123".getBytes(), amqpProperties));
 		assertEquals(String.class, result.getClass());
 	}
+
 }
diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml
index 1f1383368b..c714d89fa8 100644
--- a/src/reference/docbook/amqp.xml
+++ b/src/reference/docbook/amqp.xml
@@ -280,7 +280,11 @@ this list can also be simple patterns to be matched against the header names (e.
                                exchange-name=""]]><co               id="amqp-outbound-channel-adapter-xml-4-co" linkends="amqp-outbound-channel-adapter-xml-4" /><![CDATA[
                                order="1"]]><co                      id="amqp-outbound-channel-adapter-xml-5-co" linkends="amqp-outbound-channel-adapter-xml-5" /><![CDATA[
                                routing-key=""]]><co                 id="amqp-outbound-channel-adapter-xml-6-co" linkends="amqp-outbound-channel-adapter-xml-6" /><![CDATA[
-                               routing-key-expression=""]]><co      id="amqp-outbound-channel-adapter-xml-7-co" linkends="amqp-outbound-channel-adapter-xml-7" /><![CDATA[/>]]>
+                               routing-key-expression=""]]><co      id="amqp-outbound-channel-adapter-xml-7-co" linkends="amqp-outbound-channel-adapter-xml-7" /><![CDATA[
+                               confirm-correlation-expression=""]]><co id="amqp-outbound-channel-adapter-xml-8-co" linkends="amqp-outbound-channel-adapter-xml-8" /><![CDATA[
+                               confirm-ack-channel=""]]><co         id="amqp-outbound-channel-adapter-xml-9-co" linkends="amqp-outbound-channel-adapter-xml-9" /><![CDATA[
+                               confirm-nack-channel=""]]><co        id="amqp-outbound-channel-adapter-xml-10-co" linkends="amqp-outbound-channel-adapter-xml-10" /><![CDATA[
+                               return-channel=""]]><co              id="amqp-outbound-channel-adapter-xml-11-co" linkends="amqp-outbound-channel-adapter-xml-11" /><![CDATA[/>]]>
     </programlisting>
     <para>
         <calloutlist>
@@ -321,6 +325,33 @@ this list can also be simple patterns to be matched against the header names (e.
 					  'payload.key'). By default, this will be an empty String.
                 <emphasis>Optional</emphasis>.</para>
             </callout>
+            <callout arearefs="amqp-outbound-channel-adapter-xml-8-co" id="amqp-outbound-channel-adapter-xml-8">
+                <para>An expression defining correlation data. When provided, this configures the underlying
+                amqp template to receive publisher confirms. Requires a RabbitTemplate and a
+                CachingConnectionFactory with publisherConfirms enabled. When a publisher confirm
+                is received, it is written to either the confirm-ack-channel, or the confirm-nack-channel,
+                depending on the confirmation type. The payload of the confirm is the correlation data as
+                defined by this expression and the message will have a header 'amqp_publishConfirm' set to
+                true (ack) or false (nack). Examples: "headers['myCorrelationData']", "payload".
+                <emphasis>Optional</emphasis>.</para>
+            </callout>
+            <callout arearefs="amqp-outbound-channel-adapter-xml-9-co" id="amqp-outbound-channel-adapter-xml-9">
+                <para>The channel to which positive (ack) publisher confirms are sent; payload is
+                the correlation data defined by the <emphasis>confirm-correlation-expression</emphasis>.
+                <emphasis>Optional, default=nullChannel</emphasis>.</para>
+            </callout>
+            <callout arearefs="amqp-outbound-channel-adapter-xml-10-co" id="amqp-outbound-channel-adapter-xml-10">
+                <para>The channel to which negative (nack) publisher confirms are sent; payload is
+                the correlation data defined by the <emphasis>confirm-correlation-expression</emphasis>.
+                <emphasis>Optional, default=nullChannel</emphasis>.</para>
+            </callout>
+            <callout arearefs="amqp-outbound-channel-adapter-xml-11-co" id="amqp-outbound-channel-adapter-xml-11">
+                <para>The channel to which returned messages are sent. When provided, the underlying amqp template
+                is configured to return undeliverable messages. The message will be constructed from the
+                data received from amqp, with the following additional headers: <emphasis>amqp_returnReplyCode,
+                amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey</emphasis>.
+                <emphasis>Optional</emphasis>.</para>
+            </callout>
         </calloutlist>                                    
     </para>   
   </section>
@@ -387,8 +418,9 @@ this list can also be simple patterns to be matched against the header names (e.
                            exchange-name=""]]><co                   id="amqp-outbound-gateway-adapter-xml-4-co" linkends="amqp-outbound-gateway-adapter-xml-4" /><![CDATA[
                            order="1"]]><co                          id="amqp-outbound-gateway-adapter-xml-5-co" linkends="amqp-outbound-gateway-adapter-xml-5" /><![CDATA[
                            reply-channel=""]]><co                   id="amqp-outbound-gateway-adapter-xml-6-co" linkends="amqp-outbound-gateway-adapter-xml-6" /><![CDATA[
-                           routing-key=""]]><co                     id="amqp-outbound-gateway-adapter-xml-7-co" linkends="amqp-outbound-gateway-adapter-xml-7" /><![CDATA[                               
-                           routing-key-expression=""]]><co          id="amqp-outbound-gateway-adapter-xml-8-co" linkends="amqp-outbound-gateway-adapter-xml-8" /><![CDATA[/>]]>
+                           routing-key=""]]><co                     id="amqp-outbound-gateway-adapter-xml-7-co" linkends="amqp-outbound-gateway-adapter-xml-7" /><![CDATA[
+                           routing-key-expression=""]]><co          id="amqp-outbound-gateway-adapter-xml-8-co" linkends="amqp-outbound-gateway-adapter-xml-8" /><![CDATA[
+                           return-channel=""]]><co                  id="amqp-outbound-gateway-adapter-xml-9-co" linkends="amqp-outbound-gateway-adapter-xml-9" /><![CDATA[/>]]>
       </programlisting>
       <para>
           <calloutlist>
@@ -434,8 +466,25 @@ this list can also be simple patterns to be matched against the header names (e.
                         By default, this will be an empty String.
                   <emphasis>Optional</emphasis>.</para>
               </callout>
-          </calloutlist>                                    
-      </para> 
+              <callout arearefs="amqp-outbound-gateway-adapter-xml-9-co" id="amqp-outbound-gateway-adapter-xml-9">
+                <para>The channel to which returned messages are sent. When provided, the underlying amqp template
+                is configured to return undeliverable messages. The message will be constructed from the
+                data received from amqp, with the following additional headers: <emphasis>amqp_returnReplyCode,
+                amqp_returnReplyText, amqp_returnExchange, amqp_returnRoutingKey</emphasis>.
+                <emphasis>Optional</emphasis>.</para>
+              </callout>
+          </calloutlist>
+      </para>
+      <note>
+        <para>Prior to Spring Integration 2.2, and Spring AMQP 1.1, the outbound gateway used a new, temporary,
+        reply queue for each request. This is still the default, but now the RabbitTemplate can be configured
+        with a specific queue for replies; headers are added to the outbound message for request/reply correlation.
+        It is important that the consuming application returns these headers unchanged. The headers are
+        <classname>spring_reply_correlation</classname> and <classname>spring_reply_to</classname>. If
+        the consuming application is a Spring Integration application, these headers will be managed
+        automatically, including the case where that application might send a request/reply to a third
+        application using an outbound gateway.</para>
+      </note>
     
   </section>
 
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml
index 53ffdb35c3..a35ae01fa7 100644
--- a/src/reference/docbook/whats-new.xml
+++ b/src/reference/docbook/whats-new.xml
@@ -18,6 +18,20 @@
               Spring Integration now uses Spring 3.1.
             </para>
         </section>
+        <section id="2.2-amqp-11">
+            <title>Spring-AMQP 1.1</title>
+            <para>
+              Spring Integration now uses Spring AMQP 1.1. This enables several features
+              to be used within a Spring Integration application, including...
+            </para>
+            <itemizedlist>
+              <listitem>A fixed reply queue for the outbound gateway</listitem>
+              <listitem>HA (mirrored) queues</listitem>
+              <listitem>Publisher Confirms</listitem>
+              <listitem>Returned Messages</listitem>
+              <listitem>Support for Dead Letter Exchanges/Dead Letter Queues</listitem>
+            </itemizedlist>
+        </section>
     </section>
 
     <section id="2.2-new-components">