INT-3975: AMQP Channels: Support extract-payload

JIRA: https://jira.spring.io/browse/INT-3975

Also, switch the default mapping to map all inbound headers, but
don't map `x-*` on outbound (by default).

This will provide apps access to important headers such as `x-death` by
default, while not propagating dangerous headers outbound.

Polishing - PR Comments

Simple code style polishing

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -17,6 +17,11 @@
 package org.springframework.integration.amqp.channel;
 
 import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.amqp.support.MappingUtils;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -29,11 +34,48 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 
 	private final AmqpTemplate amqpTemplate;
 
+	private final RabbitTemplate rabbitTemplate;
+
+	private final AmqpHeaderMapper outboundHeaderMapper;
+
+	private final AmqpHeaderMapper inboundHeaderMapper;
+
+	private volatile boolean extractPayload;
+
 	private volatile boolean loggingEnabled = true;
 
+	private MessageDeliveryMode defaultDeliveryMode;
+
+	/**
+	 * Construct an instance with the supplied template and default header mappers
+	 * used if the template is a {@link RabbitTemplate} and the message is mapped.
+	 * @param amqpTemplate the template.
+	 * @see #setExtractPayload(boolean)
+	 */
 	AbstractAmqpChannel(AmqpTemplate amqpTemplate) {
+		this(amqpTemplate, DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper());
+	}
+
+	/**
+	 * Construct an instance with the supplied template and header mappers, used
+	 * when the message is mapped.
+	 * @param amqpTemplate the template.
+	 * @param outboundMapper the outbound mapper.
+	 * @param inboundMapper the inbound mapper.
+	 * @see #setExtractPayload(boolean)
+	 * @since 4.3
+	 */
+	AbstractAmqpChannel(AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
 		Assert.notNull(amqpTemplate, "amqpTemplate must not be null");
 		this.amqpTemplate = amqpTemplate;
+		if (amqpTemplate instanceof RabbitTemplate) {
+			this.rabbitTemplate = (RabbitTemplate) amqpTemplate;
+		}
+		else {
+			this.rabbitTemplate = null;
+		}
+		this.outboundHeaderMapper = outboundMapper;
+		this.inboundHeaderMapper = inboundMapper;
 	}
 
 	@Override
@@ -46,6 +88,41 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 		this.loggingEnabled = loggingEnabled;
 	}
 
+	/**
+	 * Set the delivery mode to use if the message has no
+	 * {@value org.springframework.amqp.support.AmqpHeaders#DELIVERY_MODE}
+	 * header and the message property was not set by the {@code MessagePropertiesConverter}.
+	 * @param defaultDeliveryMode the default delivery mode.
+	 * @since 4.3
+	 */
+	public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
+		this.defaultDeliveryMode = defaultDeliveryMode;
+	}
+
+	/**
+	 * Set to true to extract the payload and map the headers; otherwise
+	 * the entire message is converted and sent. Default false.
+	 * @param extractPayload true to extract and map.
+	 * @since 4.3
+	 */
+	public void setExtractPayload(boolean extractPayload) {
+		if (extractPayload) {
+			Assert.isTrue(this.rabbitTemplate != null, "amqpTemplate must be a RabbitTemplate for 'extractPayload'");
+			Assert.state(this.outboundHeaderMapper != null && this.inboundHeaderMapper != null,
+					"'extractPayload' requires both inbound and outbound header mappers");
+		}
+		this.extractPayload = extractPayload;
+	}
+
+	/**
+	 * @return the extract payload.
+	 * @see #setExtractPayload(boolean)
+	 * @since 4.3
+	 */
+	protected boolean isExtractPayload() {
+		return this.extractPayload;
+	}
+
 	/**
 	 * Subclasses may override this method to return an Exchange name.
 	 * By default, Messages will be sent to the no-name Direct Exchange.
@@ -66,13 +143,27 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel {
 		return "";
 	}
 
-	AmqpTemplate getAmqpTemplate() {
+	protected AmqpHeaderMapper getInboundHeaderMapper() {
+		return this.inboundHeaderMapper;
+	}
+
+	protected AmqpTemplate getAmqpTemplate() {
 		return this.amqpTemplate;
 	}
 
+	protected RabbitTemplate getRabbitTemplate() {
+		return this.rabbitTemplate;
+	}
+
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
-		this.amqpTemplate.convertAndSend(this.getExchangeName(), this.getRoutingKey(), message);
+		if (this.extractPayload) {
+			this.amqpTemplate.send(getExchangeName(), getRoutingKey(), MappingUtils.mapMessage(message,
+					this.rabbitTemplate.getMessageConverter(), this.outboundHeaderMapper, this.defaultDeliveryMode));
+		}
+		else {
+			this.amqpTemplate.convertAndSend(getExchangeName(), getRoutingKey(), message);
+		}
 		return true;
 	}
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.amqp.channel;
 
+import java.util.Map;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -31,9 +33,11 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.MessageDispatchingException;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.context.IntegrationProperties;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.MessageDispatcher;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageDeliveryException;
@@ -65,11 +69,44 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 
 	private final ConnectionFactory connectionFactory;
 
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template.
+	 * @see #setExtractPayload(boolean)
+	 */
 	protected AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
 			AmqpTemplate amqpTemplate) {
 		this(channelName, container, amqpTemplate, false);
 	}
 
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template.
+	 * @param outboundMapper the outbound mapper.
+	 * @param inboundMapper the inbound mapper.
+	 * @see #setExtractPayload(boolean)
+	 * @since 4.3
+	 */
+	protected AbstractSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+			AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
+		this(channelName, container, amqpTemplate, false, outboundMapper, inboundMapper);
+	}
+
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template.
+	 * @param isPubSub true for a pub/sub channel.
+	 * @see #setExtractPayload(boolean)
+	 */
 	protected AbstractSubscribableAmqpChannel(String channelName,
 			SimpleMessageListenerContainer container,
 			AmqpTemplate amqpTemplate, boolean isPubSub) {
@@ -83,6 +120,32 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 		this.admin = new RabbitAdmin(this.connectionFactory);
 	}
 
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template.
+	 * @param isPubSub true for a pub/sub channel.
+	 * @param outboundMapper the outbound mapper.
+	 * @param inboundMapper the inbound mapper.
+	 * @see #setExtractPayload(boolean)
+	 * @since 4.3
+	 */
+	protected AbstractSubscribableAmqpChannel(String channelName,
+			SimpleMessageListenerContainer container,
+			AmqpTemplate amqpTemplate, boolean isPubSub,
+			AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
+		super(amqpTemplate, outboundMapper, inboundMapper);
+		Assert.notNull(container, "container must not be null");
+		Assert.hasText(channelName, "channel name must not be empty");
+		this.channelName = channelName;
+		this.container = container;
+		this.isPubSub = isPubSub;
+		this.connectionFactory = container.getConnectionFactory();
+		this.admin = new RabbitAdmin(this.connectionFactory);
+	}
+
 	/**
 	 * Specify the maximum number of subscribers supported by the
 	 * channel's dispatcher (if it is an {@link AbstractDispatcher}).
@@ -131,7 +194,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 				: new SimpleMessageConverter();
 		MessageListener listener = new DispatchingMessageListener(converter,
 				this.dispatcher, this, this.isPubSub,
-				this.getMessageBuilderFactory());
+				getMessageBuilderFactory(), getInboundHeaderMapper());
 		this.container.setMessageListener(listener);
 		if (!this.container.isActive()) {
 			this.container.afterPropertiesSet();
@@ -157,9 +220,11 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 
 		private final MessageBuilderFactory messageBuilderFactory;
 
+		private final AmqpHeaderMapper inboundHeaderMapper;
+
 		private DispatchingMessageListener(MessageConverter converter,
 				MessageDispatcher dispatcher, AbstractSubscribableAmqpChannel channel, boolean isPubSub,
-				MessageBuilderFactory messageBuilderFactory) {
+				MessageBuilderFactory messageBuilderFactory, AmqpHeaderMapper inboundHeaderMapper) {
 			Assert.notNull(converter, "MessageConverter must not be null");
 			Assert.notNull(dispatcher, "MessageDispatcher must not be null");
 			this.converter = converter;
@@ -167,6 +232,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 			this.channel = channel;
 			this.isPubSub = isPubSub;
 			this.messageBuilderFactory = messageBuilderFactory;
+			this.inboundHeaderMapper = inboundHeaderMapper;
 		}
 
 
@@ -177,7 +243,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 				Object converted = this.converter.fromMessage(message);
 				if (converted != null) {
 					messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
-							: this.messageBuilderFactory.withPayload(converted).build();
+							: buildMessage(message, converted);
 					this.dispatcher.dispatch(messageToSend);
 				}
 				else if (this.logger.isWarnEnabled()) {
@@ -203,6 +269,18 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 						"while attempting to convert and dispatch Message.", e);
 			}
 		}
+
+		protected Message<Object> buildMessage(org.springframework.amqp.core.Message message, Object converted) {
+			AbstractIntegrationMessageBuilder<Object> messageBuilder =
+					this.messageBuilderFactory.withPayload(converted);
+			if (this.channel.isExtractPayload()) {
+				Map<String, Object> headers =
+						this.inboundHeaderMapper.toHeadersFromRequest(message.getMessageProperties());
+				messageBuilder.copyHeaders(headers);
+			}
+			return messageBuilder.build();
+		}
+
 	}
 
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -20,6 +20,7 @@ import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
 import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -34,12 +35,37 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq
 	private volatile String queueName;
 
 
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template.
+	 * @see #setExtractPayload(boolean)
+	 */
 	public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
 			AmqpTemplate amqpTemplate) {
 		super(channelName, container, amqpTemplate);
 	}
 
 
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template.
+	 * @param outboundMapper the outbound mapper.
+	 * @param inboundMapper the inbound mapper.
+	 * @see #setExtractPayload(boolean)
+	 * @since 4.3
+	 */
+	public PointToPointSubscribableAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+			AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
+		super(channelName, container, amqpTemplate, outboundMapper, inboundMapper);
+	}
+
+
 	/**
 	 * Provide a Queue name to be used. If this is not provided,
 	 * the Queue's name will be the same as the channel name.

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -19,12 +19,14 @@ package org.springframework.integration.amqp.channel;
 import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
 import org.springframework.integration.support.management.PollableChannelManagement;
 import org.springframework.messaging.Message;
@@ -54,12 +56,35 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 
 	private volatile int executorInterceptorsSize;
 
+	/**
+	 * Construct an instance with the supplied name, template and default header mappers
+	 * used if the template is a {@link RabbitTemplate} and the message is mapped.
+	 * @param channelName the channel name.
+	 * @param amqpTemplate the template.
+	 * @see #setExtractPayload(boolean)
+	 */
 	public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate) {
 		super(amqpTemplate);
 		Assert.hasText(channelName, "channel name must not be empty");
 		this.channelName = channelName;
 	}
 
+	/**
+	 * Construct an instance with the supplied name, template and header mappers.
+	 * @param channelName the channel name.
+	 * @param amqpTemplate the template.
+	 * @param outboundMapper the outbound mapper.
+	 * @param inboundMapper the inbound mapper.
+	 * @see #setExtractPayload(boolean)
+	 * @since 4.3
+	 */
+	public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper,
+			AmqpHeaderMapper inboundMapper) {
+		super(amqpTemplate, inboundMapper, outboundMapper);
+		Assert.hasText(channelName, "channel name must not be empty");
+		this.channelName = channelName;
+	}
+
 
 	/**
 	 * Provide an explicitly configured queue name. If this is not provided, then a Queue will be created
@@ -141,7 +166,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 					 return null;
 				}
 			}
-			Object object = getAmqpTemplate().receiveAndConvert(this.queueName);
+			Object object = doReceive();
 			if (object == null) {
 				if (isLoggingEnabled() && logger.isTraceEnabled()) {
 					logger.trace("postReceive on channel '" + this + "', message is null");
@@ -179,6 +204,29 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 		}
 	}
 
+
+	protected Object doReceive() {
+		if (!isExtractPayload()) {
+			return getAmqpTemplate().receiveAndConvert(this.queueName);
+		}
+		else {
+			RabbitTemplate rabbitTemplate = getRabbitTemplate();
+			org.springframework.amqp.core.Message message = rabbitTemplate.receive(this.queueName);
+			if (message != null) {
+				Object payload = rabbitTemplate.getMessageConverter().fromMessage(message);
+				Map<String, Object> headers = getInboundHeaderMapper()
+						.toHeadersFromRequest(message.getMessageProperties());
+				return getMessageBuilderFactory()
+						.withPayload(payload)
+						.copyHeaders(headers)
+						.build();
+			}
+			else {
+				return null;
+			}
+		}
+	}
+
 	@Override
 	public Message<?> receive(long timeout) {
 		if (isLoggingEnabled() && logger.isInfoEnabled()) {

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -28,6 +28,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionListener;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
 
@@ -46,11 +47,35 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 
 	private volatile boolean initialized;
 
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template.
+	 * @see #setExtractPayload(boolean)
+	 */
 	public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container,
 			AmqpTemplate amqpTemplate) {
 		super(channelName, container, amqpTemplate, true);
 	}
 
+	/**
+	 * Construct an instance with the supplied name, container and template; default header
+	 * mappers will be used if the message is mapped.
+	 * @param channelName the channel name.
+	 * @param container the container.
+	 * @param amqpTemplate the template
+	 * @param outboundMapper the outbound mapper.
+	 * @param inboundMapper the inbound mapper.
+	 * @see #setExtractPayload(boolean)
+	 * @since 4.3
+	 */
+	public PublishSubscribeAmqpChannel(String channelName, SimpleMessageListenerContainer container,
+			AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) {
+		super(channelName, container, amqpTemplate, true, outboundMapper, inboundMapper);
+	}
+
 
 	/**
 	 * Configure the FanoutExchange instance. If this is not provided, then a

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -41,6 +42,8 @@ import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
 import org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel;
 import org.springframework.integration.amqp.channel.PollableAmqpChannel;
 import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.interceptor.TransactionAttribute;
@@ -124,6 +127,13 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private volatile Boolean missingQueuesFatal;
 
+	private volatile MessageDeliveryMode defaultDeliveryMode;
+
+	private volatile Boolean extractPayload;
+
+	private volatile AmqpHeaderMapper outboundHeaderMapper = DefaultAmqpHeaderMapper.outboundMapper();
+
+	private volatile AmqpHeaderMapper inboundHeaderMapper = DefaultAmqpHeaderMapper.inboundMapper();
 
 	public AmqpChannelFactoryBean() {
 		this(true);
@@ -307,6 +317,22 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		this.missingQueuesFatal = missingQueuesFatal;
 	}
 
+	public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
+		this.defaultDeliveryMode = defaultDeliveryMode;
+	}
+
+	public void setExtractPayload(Boolean extractPayload) {
+		this.extractPayload = extractPayload;
+	}
+
+	public void setOutboundHeaderMapper(AmqpHeaderMapper outboundMapper) {
+		this.outboundHeaderMapper = outboundMapper;
+	}
+
+	public void setInboundHeaderMapper(AmqpHeaderMapper inboundMapper) {
+		this.inboundHeaderMapper = inboundMapper;
+	}
+
 	@Override
 	public Class<?> getObjectType() {
 		return (this.channel != null) ? this.channel.getClass() : AbstractAmqpChannel.class;
@@ -321,7 +347,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 			}
 			if (this.isPubSub) {
 				PublishSubscribeAmqpChannel pubsub = new PublishSubscribeAmqpChannel(
-						this.beanName, container, this.amqpTemplate);
+						this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper, this.inboundHeaderMapper);
 				if (this.exchange != null) {
 					pubsub.setExchange(this.exchange);
 				}
@@ -332,7 +358,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 			}
 			else {
 				PointToPointSubscribableAmqpChannel p2p = new PointToPointSubscribableAmqpChannel(
-						this.beanName, container, this.amqpTemplate);
+						this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper, this.inboundHeaderMapper);
 				if (StringUtils.hasText(this.queueName)) {
 					p2p.setQueueName(this.queueName);
 				}
@@ -344,7 +370,8 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		}
 		else {
 			Assert.isTrue(!this.isPubSub, "An AMQP 'publish-subscribe-channel' must be message-driven.");
-			PollableAmqpChannel pollable = new PollableAmqpChannel(this.beanName, this.amqpTemplate);
+			PollableAmqpChannel pollable = new PollableAmqpChannel(this.beanName, this.amqpTemplate,
+					this.outboundHeaderMapper, this.inboundHeaderMapper);
 			if (this.amqpAdmin != null) {
 				pollable.setAmqpAdmin(this.amqpAdmin);
 			}
@@ -360,6 +387,12 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 		if (this.getBeanFactory() != null) {
 			this.channel.setBeanFactory(this.getBeanFactory());
 		}
+		if (this.defaultDeliveryMode != null) {
+			this.channel.setDefaultDeliveryMode(this.defaultDeliveryMode);
+		}
+		if (this.extractPayload != null) {
+			this.channel.setExtractPayload(this.extractPayload);
+		}
 		this.channel.afterPropertiesSet();
 		return this.channel;
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -75,6 +75,10 @@ public class AmqpChannelParser extends AbstractChannelParser {
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-attribute");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "transaction-manager");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "tx-size");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-delivery-mode");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "outbound-header-mapper");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "inbound-header-mapper");
 		return builder;
 	}
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java

@@ -20,12 +20,10 @@ import java.util.HashMap;
 import java.util.Map;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
-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.support.CorrelationData;
 import org.springframework.amqp.support.AmqpHeaders;
-import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.context.Lifecycle;
@@ -367,28 +365,6 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
 		return routingKey;
 	}
 
-	protected org.springframework.amqp.core.Message mapMessage(Message<?> requestMessage, MessageConverter converter) {
-		MessageProperties amqpMessageProperties = new MessageProperties();
-		org.springframework.amqp.core.Message amqpMessage;
-		if (converter instanceof ContentTypeDelegatingMessageConverter) {
-			getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
-			amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
-		}
-		else { // See INT-3002 - map headers last if we're not using a CTDMC
-			amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
-			getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
-		}
-		checkDeliveryMode(requestMessage, amqpMessageProperties);
-		return amqpMessage;
-	}
-
-	private void checkDeliveryMode(Message<?> requestMessage, MessageProperties messageProperties) {
-		if (this.defaultDeliveryMode != null &&
-				requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) {
-			messageProperties.setDeliveryMode(this.defaultDeliveryMode);
-		}
-	}
-
 	protected Message<?> buildReplyMessage(MessageConverter converter,
 			org.springframework.amqp.core.Message amqpReplyMessage) {
 		Object replyObject = converter.fromMessage(amqpReplyMessage);

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java

@@ -25,6 +25,7 @@ import org.springframework.amqp.rabbit.support.CorrelationData;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.context.Lifecycle;
 import org.springframework.expression.Expression;
+import org.springframework.integration.amqp.support.MappingUtils;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
@@ -127,7 +128,8 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
 			final Message<?> requestMessage, CorrelationData correlationData) {
 		if (this.amqpTemplate instanceof RabbitTemplate) {
 			MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
-			org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter);
+			org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
+					getHeaderMapper(), getDefaultDeliveryMode());
 			((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData);
 		}
 		else {
@@ -149,7 +151,8 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
 		Assert.isInstanceOf(RabbitTemplate.class, this.amqpTemplate,
 				"RabbitTemplate implementation is required for publisher confirms");
 		MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
-		org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter);
+		org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
+				getHeaderMapper(), getDefaultDeliveryMode());
 		org.springframework.amqp.core.Message amqpReplyMessage =
 				((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage,
 						correlationData);

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java

@@ -22,6 +22,7 @@ import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
 import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture;
 import org.springframework.amqp.rabbit.support.CorrelationData;
 import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.integration.amqp.support.MappingUtils;
 import org.springframework.integration.handler.ReplyRequiredException;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
@@ -60,7 +61,9 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		RabbitMessageFuture future = this.template.sendAndReceive(generateExchangeName(requestMessage),
-				generateRoutingKey(requestMessage), mapMessage(requestMessage, this.messageConverter));
+				generateRoutingKey(requestMessage),
+				MappingUtils.mapMessage(requestMessage, this.messageConverter, getHeaderMapper(),
+						getDefaultDeliveryMode()));
 		future.addCallback(new FutureCallback(requestMessage));
 		CorrelationData correlationData = generateCorrelationData(requestMessage);
 		if (correlationData != null && future.getConfirm() != null) {

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java

@@ -119,7 +119,17 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 	 */
 	@Deprecated
 	public DefaultAmqpHeaderMapper() {
+		this(null, null);
+	}
+
+	private DefaultAmqpHeaderMapper(String[] requestHeaderNames, String[] replyHeaderNames) {
 		super(AmqpHeaders.PREFIX, STANDARD_HEADER_NAMES, STANDARD_HEADER_NAMES);
+		if (requestHeaderNames != null) {
+			setRequestHeaderNames(requestHeaderNames);
+		}
+		if (replyHeaderNames != null) {
+			setReplyHeaderNames(replyHeaderNames);
+		}
 	}
 
 	/**
@@ -411,49 +421,61 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		}
 	}
 
+	/**
+	 * Construct a default inbound header mapper.
+	 * @return the mapper.
+	 * @see #inboundRequestHeaders()
+	 * @see #inboundReplyHeaders()
+	 * @since 4.3
+	 */
 	public static DefaultAmqpHeaderMapper inboundMapper() {
-		DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
-		mapper.setRequestHeaderNames(inboundRequestHeaders());
-		mapper.setReplyHeaderNames(inboundReplyHeaders());
-		return mapper;
+		return new DefaultAmqpHeaderMapper(inboundRequestHeaders(), inboundReplyHeaders());
 	}
 
+	/**
+	 * Construct a default outbound header mapper.
+	 * @return the mapper.
+	 * @see #outboundRequestHeaders()
+	 * @see #outboundReplyHeaders()
+	 * @since 4.3
+	 */
 	public static DefaultAmqpHeaderMapper outboundMapper() {
-		DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
-		mapper.setRequestHeaderNames(outboundRequestHeaders());
-		mapper.setReplyHeaderNames(outboundReplyHeaders());
-		return mapper;
+		return new DefaultAmqpHeaderMapper(outboundRequestHeaders(), outboundReplyHeaders());
 	}
 
 	/**
 	 * @return the default request headers for an inbound mapper.
+	 * @since 4.3
 	 */
 	public static String[] inboundRequestHeaders() {
-		return safeInboundHeaders();
+		return new String[] { "*" };
 	}
 
 	/**
 	 * @return the default reply headers for an inbound mapper.
+	 * @since 4.3
 	 */
 	public static String[] inboundReplyHeaders() {
-		return new String[] { "*" };
+		return safeOutboundHeaders();
 	}
 
 	/**
 	 * @return the default request headers for an outbound mapper.
+	 * @since 4.3
 	 */
 	public static String[] outboundRequestHeaders() {
-		return new String[] { "*" };
+		return safeOutboundHeaders();
 	}
 
 	/**
 	 * @return the default reply headers for an outbound mapper.
+	 * @since 4.3
 	 */
 	public static String[] outboundReplyHeaders() {
-		return safeInboundHeaders();
+		return new String[] { "*" };
 	}
 
-	private static String[] safeInboundHeaders() {
+	private static String[] safeOutboundHeaders() {
 		return new String[] { "!x-*", "*" };
 	}
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/MappingUtils.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2016 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.integration.amqp.support;
+
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.messaging.Message;
+
+/**
+ * Utility methods used during message mapping.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public class MappingUtils {
+
+	/**
+	 * Map an o.s.Message to an o.s.a.core.Message.
+	 * @param requestMessage the request message.
+	 * @param converter the message converter to use.
+	 * @param headerMapper the header mapper to use.
+	 * @param defaultDeliveryMode the default delivery mode.
+	 * @return the mapped Message.
+	 */
+	public static org.springframework.amqp.core.Message mapMessage(Message<?> requestMessage,
+			MessageConverter converter, AmqpHeaderMapper headerMapper, MessageDeliveryMode defaultDeliveryMode) {
+		MessageProperties amqpMessageProperties = new MessageProperties();
+		org.springframework.amqp.core.Message amqpMessage;
+		if (converter instanceof ContentTypeDelegatingMessageConverter) {
+			headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
+			amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
+		}
+		else { // See INT-3002 - map headers last if we're not using a CTDMC
+			amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
+			headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
+		}
+		checkDeliveryMode(requestMessage, amqpMessageProperties, defaultDeliveryMode);
+		return amqpMessage;
+	}
+
+	/**
+	 * Check the delivery mode and update with the default if not already present.
+	 * @param requestMessage the request message.
+	 * @param messageProperties the mapped message properties.
+	 * @param defaultDeliveryMode the default delivery mode.
+	 */
+	public static void checkDeliveryMode(Message<?> requestMessage, MessageProperties messageProperties,
+			MessageDeliveryMode defaultDeliveryMode) {
+		if (defaultDeliveryMode != null &&
+				requestMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE) == null) {
+			messageProperties.setDeliveryMode(defaultDeliveryMode);
+		}
+	}
+
+}

spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.3.xsd

@@ -344,6 +344,45 @@
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="default-delivery-mode">
+			<xsd:annotation>
+				<xsd:documentation>
+	The default delivery mode for messages; 'PERSISTENT' or 'NON_PERSISTENT'. Overridden if the 'header-mapper'
+	sets the delivery mode. The 'DefaultHeaderMapper' sets the value if the
+	Spring Integration message header 'amqp_deliveryMode' is present. If this attribute is not supplied and
+	the header mapper doesn't set it, the default depends on the underlying spring-amqp 'MessagePropertiesConverter'
+	used by the 'RabbitTemplate'. If that is not customized at all, the default is 'PERSISTENT'.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="deliveryModeEnumeration xsd:string" />
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="extract-payload">
+			<xsd:annotation>
+				<xsd:documentation>
+	Set to 'true' to extract the message payload and map the o.s.messaging.Message to an o.s.amqp.core.Message in
+	a similar manner to a pair of channel adapters. When 'false' the entire message is converted requiring either
+	Java serializable contents or a custom message converter. Also see inbound and outbound mapped headers.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="outbound-header-mapper">
+			<xsd:annotation>
+				<xsd:documentation>
+	The header mapper to use when sending and 'extract-payload' is true. The default mapper is
+	DefaultAmqpHeaderMapper.outboundMapper() which maps all headers except 'x-*'.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="inbound-header-mapper">
+			<xsd:annotation>
+				<xsd:documentation>
+	The header mapper to use when receiving and 'extract-payload' is true. The default mapper is
+	DefaultAmqpHeaderMapper.inboundMapper() which maps all headers.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attributeGroup ref="containerAndTemplateAttributes"/>
 		<xsd:attributeGroup ref="integration:subscribersAttributeGroup" />
 	</xsd:complexType>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests-context.xml

@@ -9,8 +9,24 @@
 		http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
 
-	<int-amqp:publish-subscribe-channel id="foo" />
+	<int-amqp:publish-subscribe-channel id="channel" />
 
 	<rabbit:connection-factory id="rabbitConnectionFactory" host="localhost" />
 
+	<int-amqp:channel id="withEP" extract-payload="true" message-converter="jackson" />
+
+	<int-amqp:channel id="pollableWithEP" extract-payload="true" message-driven="false" message-converter="jackson" />
+
+	<int-amqp:publish-subscribe-channel id="pubSubWithEP" extract-payload="true" message-converter="jackson" />
+
+	<int:bridge input-channel="withEP" output-channel="out" />
+
+	<int:bridge input-channel="pubSubWithEP" output-channel="out" />
+
+	<int:channel id="out">
+		<int:queue />
+	</int:channel>
+
+	<bean id="jackson" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
+
 </beans>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2015 the original author or authors.
+ * Copyright 2014-2016 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.amqp.channel;
 
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -43,11 +44,13 @@ import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
 import org.springframework.integration.amqp.rule.BrokerRunning;
+import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.support.LogAdjustingTestSupport;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.annotation.DirtiesContext.ClassMode;
@@ -66,11 +69,24 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 public class ChannelTests extends LogAdjustingTestSupport {
 
 	@ClassRule
-	public static final BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+	public static final BrokerRunning brokerIsRunning =
+		BrokerRunning.isRunningWithEmptyQueues("pollableWithEP", "withEP");
 
 	@Autowired
 	private PublishSubscribeAmqpChannel channel;
 
+	@Autowired
+	private PollableAmqpChannel pollableWithEP;
+
+	@Autowired
+	private PointToPointSubscribableAmqpChannel withEP;
+
+	@Autowired
+	private PublishSubscribeAmqpChannel pubSubWithEP;
+
+	@Autowired
+	private PollableChannel out;
+
 	@Autowired
 	private CachingConnectionFactory factory;
 
@@ -81,9 +97,11 @@ public class ChannelTests extends LogAdjustingTestSupport {
 	@After
 	public void tearDown() {
 		new RabbitAdmin(this.factory).deleteExchange("si.fanout.foo");
+		brokerIsRunning.removeTestQueues();
 	}
 
 	@Test
+	@DirtiesContext
 	public void pubSubLostConnectionTest() throws Exception {
 		final CyclicBarrier latch = new CyclicBarrier(2);
 		channel.subscribe(new MessageHandler() {
@@ -107,6 +125,7 @@ public class ChannelTests extends LogAdjustingTestSupport {
 		this.channel.send(new GenericMessage<String>("bar"));
 		latch.await(10, TimeUnit.SECONDS);
 		this.channel.destroy();
+		this.pubSubWithEP.destroy();
 		assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size());
 	}
 
@@ -179,4 +198,84 @@ public class ChannelTests extends LogAdjustingTestSupport {
 		rabbitAdmin.deleteExchange("si.fanout.testChannel");
 	}
 
+	@Test
+	public void extractPayloadTests() throws Exception {
+		Foo foo = new Foo("bar");
+		Message<?> message = MessageBuilder.withPayload(foo).setHeader("baz", "qux").build();
+		this.pollableWithEP.send(message);
+		Message<?> received = this.pollableWithEP.receive();
+		int n = 0;
+		while (received == null && n++ < 100) {
+			Thread.sleep(100);
+			received = this.pollableWithEP.receive();
+		}
+		assertNotNull(received);
+		assertThat((Foo) received.getPayload(), equalTo(foo));
+		assertThat((String) received.getHeaders().get("baz"), equalTo("qux"));
+
+		this.withEP.send(message);
+		received = this.out.receive(10000);
+		assertNotNull(received);
+		assertThat((Foo) received.getPayload(), equalTo(foo));
+		assertThat((String) received.getHeaders().get("baz"), equalTo("qux"));
+
+		this.pubSubWithEP.send(message);
+		received = this.out.receive(10000);
+		assertNotNull(received);
+		assertThat((Foo) received.getPayload(), equalTo(foo));
+		assertThat((String) received.getHeaders().get("baz"), equalTo("qux"));
+	}
+
+	public static class Foo {
+
+		private String bar;
+
+		public Foo() {
+		}
+
+		public Foo(String bar) {
+			this.bar = bar;
+		}
+
+		public String getBar() {
+			return this.bar;
+		}
+
+		public void setBar(String bar) {
+			this.bar = bar;
+		}
+
+		@Override
+		public int hashCode() {
+			final int prime = 31;
+			int result = 1;
+			result = prime * result + ((this.bar == null) ? 0 : this.bar.hashCode());
+			return result;
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (this == obj) {
+				return true;
+			}
+			if (obj == null) {
+				return false;
+			}
+			if (getClass() != obj.getClass()) {
+				return false;
+			}
+			Foo other = (Foo) obj;
+			if (this.bar == null) {
+				if (other.bar != null) {
+					return false;
+				}
+			}
+			else if (!this.bar.equals(other.bar)) {
+				return false;
+			}
+			return true;
+		}
+
+	}
+
 }

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml

@@ -22,4 +22,21 @@
 
 	<amqp:publish-subscribe-channel id="pubSub" />
 
+	<amqp:channel id="withEP" extract-payload="true" default-delivery-mode="NON_PERSISTENT"
+		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
+
+	<amqp:channel id="pollableWithEP" extract-payload="true" message-driven="false"
+		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
+
+	<amqp:publish-subscribe-channel id="pubSubWithEP" extract-payload="true"
+		inbound-header-mapper="inMapper" outbound-header-mapper="outMapper" />
+
+	<bean id="inMapper" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.integration.amqp.support.AmqpHeaderMapper" />
+	</bean>
+
+	<bean id="outMapper" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.integration.amqp.support.AmqpHeaderMapper" />
+	</bean>
+
 </beans>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -16,9 +16,12 @@
 
 package org.springframework.integration.amqp.config;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.List;
@@ -26,8 +29,13 @@ import java.util.List;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
+import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
+import org.springframework.integration.amqp.channel.PointToPointSubscribableAmqpChannel;
+import org.springframework.integration.amqp.channel.PollableAmqpChannel;
+import org.springframework.integration.amqp.channel.PublishSubscribeAmqpChannel;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageChannel;
@@ -49,6 +57,15 @@ public class AmqpChannelParserTests {
 	@Autowired
 	private ApplicationContext context;
 
+	@Autowired
+	private PollableAmqpChannel pollableWithEP;
+
+	@Autowired
+	private PointToPointSubscribableAmqpChannel withEP;
+
+	@Autowired
+	private PublishSubscribeAmqpChannel pubSubWithEP;
+
 	@Test
 	public void interceptor() {
 		MessageChannel channel = context.getBean("channelWithInterceptor", MessageChannel.class);
@@ -73,8 +90,26 @@ public class AmqpChannelParserTests {
 		assertFalse(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class));
 		assertFalse(TestUtils.getPropertyValue(channel, "container.transactional", Boolean.class));
 		assertTrue(TestUtils.getPropertyValue(channel, "amqpTemplate.transactional", Boolean.class));
+		assertFalse(TestUtils.getPropertyValue(channel, "extractPayload", Boolean.class));
 	}
 
+	@Test
+	public void testMapping() {
+		checkExtract(this.pollableWithEP);
+		checkExtract(this.withEP);
+		checkExtract(this.pubSubWithEP);
+		assertEquals(MessageDeliveryMode.NON_PERSISTENT,
+				TestUtils.getPropertyValue(this.withEP, "defaultDeliveryMode"));
+		assertNull(TestUtils.getPropertyValue(this.pollableWithEP, "defaultDeliveryMode"));
+	}
+
+	private void checkExtract(AbstractAmqpChannel channel) {
+		assertThat(TestUtils.getPropertyValue(channel, "outboundHeaderMapper").toString(),
+				containsString("Mock for AmqpHeaderMapper"));
+		assertThat(TestUtils.getPropertyValue(channel, "inboundHeaderMapper").toString(),
+				containsString("Mock for AmqpHeaderMapper"));
+		assertTrue(TestUtils.getPropertyValue(channel, "extractPayload", Boolean.class));
+	}
 
 	private static class TestInterceptor extends ChannelInterceptorAdapter {
 	}

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java

@@ -264,24 +264,28 @@ public class DefaultAmqpHeaderMapperTests {
 		assertNull(headers.get(AmqpHeaders.DELIVERY_MODE));
 		assertEquals(MessageDeliveryMode.NON_PERSISTENT, headers.get(AmqpHeaders.RECEIVED_DELIVERY_MODE));
 		assertEquals("bar", headers.get("foo"));
-		assertNull(headers.get("x-foo"));
+		assertEquals("bar", headers.get("x-foo"));
 
+		amqpProperties = new MessageProperties();
 		headers.put(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.NON_PERSISTENT);
 		mapper.fromHeadersToReply(new MessageHeaders(headers), amqpProperties);
 		assertEquals(MessageDeliveryMode.NON_PERSISTENT, amqpProperties.getDeliveryMode());
 		assertEquals("bar", amqpProperties.getHeaders().get("foo"));
-
+		assertNull(amqpProperties.getHeaders().get("x-foo"));
 
 		mapper = DefaultAmqpHeaderMapper.outboundMapper();
 		mapper.fromHeadersToRequest(new MessageHeaders(headers), amqpProperties);
 		assertEquals(MessageDeliveryMode.NON_PERSISTENT, amqpProperties.getDeliveryMode());
 		assertEquals("bar", amqpProperties.getHeaders().get("foo"));
+		assertNull(amqpProperties.getHeaders().get("x-foo"));
 
 		amqpProperties.setReceivedDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+		amqpProperties.setHeader("x-death", "foo");
 		headers = mapper.toHeadersFromReply(amqpProperties);
 		assertEquals(MessageDeliveryMode.NON_PERSISTENT, headers.get(AmqpHeaders.RECEIVED_DELIVERY_MODE));
 		assertNull(headers.get(AmqpHeaders.DELIVERY_MODE));
 		assertEquals("bar", headers.get("foo"));
+		assertEquals("foo", headers.get("x-death"));
 	}
 
 }

src/reference/asciidoc/amqp.adoc

@@ -1142,7 +1142,8 @@ This applies to both the outbound channel adapter and gateway.
 
 There are two Message Channel implementations available.
 One is point-to-point, and the other is publish/subscribe.
-Both of these channels provide a wide range of configuration attributes for the underlying AmqpTemplate and SimpleMessageListenerContainer as you have seen on the Channel Adapters and Gateways.
+Both of these channels provide a wide range of configuration attributes for the underlying AmqpTemplate and
+SimpleMessageListenerContainer as you have seen on the Channel Adapters and Gateways.
 However, the examples we'll show here are going to have minimal configuration.
 Explore the XML schema to view the available attributes.
 
@@ -1154,7 +1155,9 @@ A point-to-point channel would look like this:
 
 Under the covers a Queue named "si.p2pChannel" would be declared, and this channel will send to that Queue (technically by sending to the no-name Direct Exchange with a routing key that matches this Queue's name).
 This channel will also register a consumer on that Queue.
-If for some reason, you want the Queue to be "pollable" instead of message-driven, then simply provide the "message-driven" flag with a value of false:
+If you want the channel to be "pollable" instead of message-driven, then simply provide the "message-driven" flag with
+a value of `false`:
+
 [source,xml]
 ----
 <int-amqp:channel id="p2pPollableChannel"  message-driven="false"/>
@@ -1166,12 +1169,35 @@ A publish/subscribe channel would look like this:
 <int-amqp:publish-subscribe-channel id="pubSubChannel"/>
 ----
 
-Under the covers a Fanout Exchange named "si.fanout.pubSubChannel" would be declared, and this channel will send to that Fanout Exchange.
-This channel will also declare a server-named exclusive, auto-delete, non-durable Queue and bind that to the Fanout Exchange while registering a consumer on that Queue to receive Messages.
+Under the covers a Fanout Exchange named "si.fanout.pubSubChannel" would be declared, and this channel will send to that
+Fanout Exchange.
+This channel will also declare a server-named exclusive, auto-delete, non-durable Queue and bind that to the Fanout
+Exchange while registering a consumer on that Queue to receive Messages.
 There is no "pollable" option for a publish-subscribe-channel; it must be message-driven.
 
-Starting with _version 4.1_ AMQP Backed Message Channels, alongside with `channel-transacted`, support `template-channel-transacted` to separate `transactional` configuration for the `AbstractMessageListenerContainer` and for the `RabbitTemplate`.
-Note, previously, the `channel-transacted` was `true` by default, now it changed to `false` as standard default value for the `AbstractMessageListenerContainer`.
+Starting with _version 4.1_ AMQP Backed Message Channels, alongside with `channel-transacted`, support
+`template-channel-transacted` to separate `transactional` configuration for the `AbstractMessageListenerContainer` and
+for the `RabbitTemplate`.
+Note, previously, the `channel-transacted` was `true` by default, now it changed to `false` as standard default value
+for the `AbstractMessageListenerContainer`.
+
+Prior to _version 4.3_, AMQP-backed channels only supported messages with `Serializable` payloads and headers.
+The entire message was converted (serialized) and sent to RabbitMQ.
+Now, you can set the `extract-payload` attribute (or `setExtractPayload()` when using Java configuration) to true.
+When this flag is `true`, the message payload is converted and the headers mapped, in a similar manner to when using
+channel adapters.
+This allows AMQP-backed channels to be used with non-serializable payloads (perhaps with another message converter
+such as the `Jackson2JsonMessageConverter`).
+The default mapped headers are discussed in <<amqp-message-headers>>.
+You can modify the mapping by providing custom mappers using the `outbound-header-mapper` and `inbound-header-mapper`
+attributes.
+You can now also specify a `default-delivery-mode`, used to set the delivery mode when there is no
+`amqp_deliveryMode` header.
+By default, Spring AMQP `MessageProperties` uses `PERSISTENT` delivery mode.
+
+IMPORTANT: Just as with other persistence-backed channels, AMQP-backed channels are intended to provide message
+persistence to avoid message loss.
+They are not intended to distribute work to other peer applications; for that purpose, use channel adapters instead.
 
 ==== Configuring with Java Configuration
 
@@ -1258,7 +1284,7 @@ properties to support that.
 Any user-defined headers within the AMQP http://docs.spring.io/spring-amqp/api/org/springframework/amqp/core/MessageProperties.html[MessageProperties] WILL
 be copied to or from an AMQP Message, unless explicitly negated by the _requestHeaderNames_ and/or
 _replyHeaderNames_ properties of the `DefaultAmqpHeaderMapper`.
-For an inbound mapper, all `x-*` headers are not mapped by default.
+For an outbound mapper, no `x-*` headers are mapped by default; see the caution below for the reason why.
 
 To override the default, and revert to the pre-4.3 behavior, use `STANDARD_REQUEST_HEADERS` and
 `STANDARD_REPLY_HEADERS` in the properties.

src/reference/asciidoc/whats-new.adoc

@@ -109,6 +109,11 @@ Spring AMQP 1.6 adds support for
 https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/[Delayed Message Exchanges].
 Header mapping now supports the headers (`amqp_delay` and `amqp_receivedDelay`) used by this feature.
 
+===== AMQP-Backed Channels
+
+AMQP-backed channels now support message mapping.
+See <<amqp-channels>> for more information.
+
 ==== Redis Changes
 
 ===== List Push/Pop Direction