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

Author: garyrussell

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 void setLoggingEnabled(boolean loggingEnabled) {
 		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 @@ protected String getRoutingKey() {
 		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.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 @@ protected AbstractSubscribableAmqpChannel(String channelName,
 		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 @@ public void onInit() throws Exception {
 				: 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,16 +220,19 @@ private static class DispatchingMessageListener implements MessageListener {
 
 		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;
 			this.dispatcher = dispatcher;
 			this.channel = channel;
 			this.isPubSub = isPubSub;
 			this.messageBuilderFactory = messageBuilderFactory;
+			this.inboundHeaderMapper = inboundHeaderMapper;
 		}
 
 
@@ -177,7 +243,7 @@ public void onMessage(org.springframework.amqp.core.Message message) {
 				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 @@ else if (this.logger.isWarnEnabled()) {
 						"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.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.