GH-51: Add headers mapping to channel adapters

Fixes spring-projects/spring-integration-aws#51

README.md

@@ -289,6 +289,9 @@ An XML variant may look like:
                    queue="foo"/>
 ````
 
+Starting with _version 2.0_, the `SqsMessageHandler` can be configured with the `HeaderMapper` to map message headers to the SQS message attributes.
+See `SqsHeaderMapper` implementation for more information and also consult with [Amazon SQS Message Attributes][] about value types and restrictions.   
+
 ### Inbound Channel Adapter
 
 The SQS Inbound Channel Adapter is a `message-driven` implementation for the `MessageProducer` and is represented with
@@ -466,6 +469,9 @@ The XML variant may look like:
 			body-expression="payload.toUpperCase()"/>
 ````
 
+Starting with _version 2.0_, the `SnsMessageHandler` can be configured with the `HeaderMapper` to map message headers to the SNS message attributes.
+See `SnsHeaderMapper` implementation for more information and also consult with [[Amazon SNS Message Attributes]][] about value types and restrictions.   
+
 ## Metadata Store for Amazon DynamoDB
 
 The `DynamoDbMetaDataStore`, a `ConcurrentMetadataStore` implementation, is provided to keep the metadata for Spring Integration components in the distributed Amazon DynamoDB store. 
@@ -530,6 +536,9 @@ When `CheckpointMode.manual` is used the `Checkpointer` instance is populated to
 The consumer group is included to the metadata store `key`.
 When records are consumed, they are filtered by the last stored `lastCheckpoint` under the key as `[CONSUMER_GROUP]:[STREAM]:[SHARD_ID]`.
 
+Starting with _version 2.0_, the `KinesisMessageDrivenChannelAdapter` can be configured with the `InboundMessageMapper` to extract message headers embedded into the record data (if any).
+See `EmbeddedJsonHeadersMessageMapper` implementation for more information.
+
 ### Outbound Channel Adapter
 
 The `KinesisMessageHandler` is an `AbstractMessageHandler` to perform put record to the Kinesis stream.
@@ -570,6 +579,9 @@ public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis,
 }
 ````
 
+Starting with _version 2.0_, the `KinesisMessageHandler` can be configured with the `OutboundMessageMapper` to embed message headers into the record data alongside with the payload.
+See `EmbeddedJsonHeadersMessageMapper` implementation for more information.
+
 For testing application with the Kinesis Channel Adapters you can use [Kinesalite][] NPM module.
 What you need in your application is to configure Kinesis client properly:
 
@@ -605,3 +617,5 @@ Also you can use for you testing purpose a copy of `org.springframework.integrat
 [Dynalite]: https://github.com/mhart/dynalite
 [Kinesis Client Library]: https://github.com/awslabs/amazon-kinesis-client
 [Kinesalite]: https://github.com/mhart/kinesalite
+[Amazon SQS Message Attributes]: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-attributes.html
+[Amazon SNS Message Attributes]: https://docs.aws.amazon.com/sns/latest/dg/SNSMessageAttributes.html

build.gradle

@@ -3,7 +3,7 @@ buildscript {
 		maven { url 'http://repo.spring.io/plugins-release' }
 	}
 	dependencies {
-		classpath 'io.spring.gradle:dependency-management-plugin:1.0.4.RELEASE'
+		classpath 'io.spring.gradle:dependency-management-plugin:1.0.5.RELEASE'
 		classpath 'io.spring.gradle:spring-io-plugin:0.0.8.RELEASE'
 	}
 }
@@ -33,9 +33,9 @@ repositories {
 ext {
 	assertjVersion = '3.9.1'
 	servletApiVersion = '3.1.0'
-	log4jVersion = '2.10.0'
+	log4jVersion = '2.11.0'
 	springCloudAwsVersion = '2.0.0.M4'
-	springIntegrationVersion = '5.0.3.RELEASE'
+	springIntegrationVersion = '5.0.4.RELEASE'
 
 	idPrefix = 'aws'
 

src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java

@@ -44,6 +44,7 @@ import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.serializer.support.DeserializingConverter;
 import org.springframework.integration.aws.support.AwsHeaders;
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.mapping.InboundMessageMapper;
 import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
@@ -136,6 +137,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i
 
 	private boolean resetCheckpoints;
 
+	private InboundMessageMapper<byte[]> embeddedHeadersMapper;
+
 	private volatile boolean active;
 
 	private volatile int consumerInvokerMaxCapacity;
@@ -258,6 +261,15 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i
 		this.idleBetweenPolls = Math.max(250, idleBetweenPolls);
 	}
 
+	/**
+	 * Specify an {@link InboundMessageMapper} to extract message headers embedded into the record data.
+	 * @param embeddedHeadersMapper the {@link InboundMessageMapper} to use.
+	 * @since 2.0
+	 */
+	public void setEmbeddedHeadersMapper(InboundMessageMapper<byte[]> embeddedHeadersMapper) {
+		this.embeddedHeadersMapper = embeddedHeadersMapper;
+	}
+
 	@Override
 	protected void onInit() {
 		super.onInit();
@@ -931,9 +943,26 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i
 						for (Record record : records) {
 							Object payload = record.getData().array();
 
-							if (KinesisMessageDrivenChannelAdapter.this.converter != null) {
+							Message<?> messageToUse = null;
+							if (KinesisMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) {
+								try {
+									messageToUse =
+											KinesisMessageDrivenChannelAdapter.this.embeddedHeadersMapper
+													.toMessage((byte[]) payload);
+
+									payload = messageToUse.getPayload();
+								}
+								catch (Exception e) {
+									logger.warn("Could not parse embedded headers. Remain payload untouched.", e);
+								}
+							}
+
+							if (payload instanceof byte[] &&
+									KinesisMessageDrivenChannelAdapter.this.converter != null) {
+
 								payload = KinesisMessageDrivenChannelAdapter.this.converter.convert((byte[]) payload);
 							}
+
 							AbstractIntegrationMessageBuilder<Object> messageBuilder = getMessageBuilderFactory()
 									.withPayload(payload)
 									.setHeader(AwsHeaders.RECEIVED_STREAM, this.shardOffset.getStream())
@@ -944,6 +973,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i
 								messageBuilder.setHeader(AwsHeaders.CHECKPOINTER, this.checkpointer);
 							}
 
+							if (messageToUse != null) {
+								messageBuilder.copyHeadersIfAbsent(messageToUse.getHeaders());
+							}
+
 							performSend(messageBuilder, record);
 
 							if (CheckpointMode.record.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) {

src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.integration.aws.support.AwsRequestFailureException;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.handler.AbstractMessageProducingHandler;
+import org.springframework.integration.mapping.HeaderMapper;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.DefaultErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageStrategy;
@@ -44,11 +45,13 @@ import com.amazonaws.handlers.AsyncHandler;
  * Utilizes common logic ({@link AsyncHandler}, {@link ErrorMessageStrategy},
  * {@code failureChannel} etc.) and message pre- and post-processing,
  *
+ * @param <H> the headers container type.
+ *
  * @author Artem Bilan
  *
  * @since 2.0
  */
-public abstract class AbstractAwsMessageHandler extends AbstractMessageProducingHandler {
+public abstract class AbstractAwsMessageHandler<H> extends AbstractMessageProducingHandler {
 
 	protected static final long DEFAULT_SEND_TIMEOUT = 10000;
 
@@ -66,6 +69,8 @@ public abstract class AbstractAwsMessageHandler extends AbstractMessageProducing
 
 	private String failureChannelName;
 
+	private HeaderMapper<H> headerMapper;
+
 	public void setAsyncHandler(AsyncHandler<? extends AmazonWebServiceRequest, ?> asyncHandler) {
 		this.asyncHandler = asyncHandler;
 	}
@@ -141,6 +146,22 @@ public abstract class AbstractAwsMessageHandler extends AbstractMessageProducing
 		return this.errorMessageStrategy;
 	}
 
+	/**
+	 * Specify a {@link HeaderMapper} to map outbound headers.
+	 * @param headerMapper the {@link HeaderMapper} to map outbound headers.
+	 */
+	public void setHeaderMapper(HeaderMapper<H> headerMapper) {
+		doSetHeaderMapper(headerMapper);
+	}
+
+	protected final void doSetHeaderMapper(HeaderMapper<H> headerMapper) {
+		this.headerMapper = headerMapper;
+	}
+
+	protected HeaderMapper<H> getHeaderMapper() {
+		return this.headerMapper;
+	}
+
 	protected EvaluationContext getEvaluationContext() {
 		return this.evaluationContext;
 	}
@@ -213,7 +234,7 @@ public abstract class AbstractAwsMessageHandler extends AbstractMessageProducing
 		};
 	}
 
-	protected abstract Future<?> handleMessageToAws(Message<?> message);
+	protected abstract Future<?> handleMessageToAws(Message<?> message) throws Exception;
 
 	protected abstract void additionalOnSuccessHeaders(AbstractIntegrationMessageBuilder<?> messageBuilder,
 			AmazonWebServiceRequest request, Object result);

src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java

@@ -25,8 +25,12 @@ import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.aws.support.AwsHeaders;
 import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.mapping.HeaderMapper;
+import org.springframework.integration.mapping.OutboundMessageMapper;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.integration.support.MutableMessage;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -50,19 +54,21 @@ import com.amazonaws.services.kinesis.model.PutRecordsResult;
  * @see AmazonKinesisAsync#putRecords(PutRecordsRequest)
  * @see com.amazonaws.handlers.AsyncHandler
  */
-public class KinesisMessageHandler extends AbstractAwsMessageHandler {
+public class KinesisMessageHandler extends AbstractAwsMessageHandler<Void> {
 
 	private final AmazonKinesisAsync amazonKinesis;
 
 	private Converter<Object, byte[]> converter = new SerializingConverter();
 
-	private volatile Expression streamExpression;
+	private Expression streamExpression;
 
-	private volatile Expression partitionKeyExpression;
+	private Expression partitionKeyExpression;
 
-	private volatile Expression explicitHashKeyExpression;
+	private Expression explicitHashKeyExpression;
 
-	private volatile Expression sequenceNumberExpression;
+	private Expression sequenceNumberExpression;
+
+	private OutboundMessageMapper<byte[]> embeddedHeadersMapper;
 
 	public KinesisMessageHandler(AmazonKinesisAsync amazonKinesis) {
 		Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null.");
@@ -123,11 +129,33 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler {
 		this.sequenceNumberExpression = sequenceNumberExpression;
 	}
 
+	/**
+	 * Specify a {@link OutboundMessageMapper} for embedding message headers into the record data
+	 * together with payload.
+	 * @param embeddedHeadersMapper the {@link OutboundMessageMapper} to embed headers into the record data.
+	 * @since 2.0
+	 * @see org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper
+	 */
+	public void setEmbeddedHeadersMapper(OutboundMessageMapper<byte[]> embeddedHeadersMapper) {
+		this.embeddedHeadersMapper = embeddedHeadersMapper;
+	}
+
+	/**
+	 * Unsupported operation. Use {@link #setEmbeddedHeadersMapper} instead.
+	 * @param headerMapper is not used.
+	 * @see #setEmbeddedHeadersMapper
+	 */
 	@Override
-	protected Future<?> handleMessageToAws(Message<?> message) {
+	public void setHeaderMapper(HeaderMapper<Void> headerMapper) {
+		throw new UnsupportedOperationException("Kinesis doesn't support headers.\n" +
+				"Consider to use 'OutboundMessageMapper<byte[]>' for embedding headers into the record data.");
+	}
+
+	@Override
+	protected Future<?> handleMessageToAws(Message<?> message) throws Exception {
 		if (message.getPayload() instanceof PutRecordsRequest) {
 			AsyncHandler<PutRecordsRequest, PutRecordsResult> asyncHandler =
-							obtainAsyncHandler(message, (PutRecordsRequest) message.getPayload());
+					obtainAsyncHandler(message, (PutRecordsRequest) message.getPayload());
 
 			return this.amazonKinesis.putRecordsAsync((PutRecordsRequest) message.getPayload(), asyncHandler);
 		}
@@ -144,8 +172,9 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler {
 		}
 	}
 
-	private PutRecordRequest buildPutRecordRequest(Message<?> message) {
-		String stream = message.getHeaders().get(AwsHeaders.STREAM, String.class);
+	private PutRecordRequest buildPutRecordRequest(Message<?> message) throws Exception {
+		MessageHeaders messageHeaders = message.getHeaders();
+		String stream = messageHeaders.get(AwsHeaders.STREAM, String.class);
 		if (!StringUtils.hasText(stream) && this.streamExpression != null) {
 			stream = this.streamExpression.getValue(getEvaluationContext(), message, String.class);
 		}
@@ -153,7 +182,7 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler {
 				"Consider configuring this handler with a 'stream'( or 'streamExpression') or supply an " +
 				"'aws_stream' message header.");
 
-		String partitionKey = message.getHeaders().get(AwsHeaders.PARTITION_KEY, String.class);
+		String partitionKey = messageHeaders.get(AwsHeaders.PARTITION_KEY, String.class);
 		if (!StringUtils.hasText(partitionKey) && this.partitionKeyExpression != null) {
 			partitionKey = this.partitionKeyExpression.getValue(getEvaluationContext(), message, String.class);
 		}
@@ -166,25 +195,38 @@ public class KinesisMessageHandler extends AbstractAwsMessageHandler {
 						? this.explicitHashKeyExpression.getValue(getEvaluationContext(), message, String.class)
 						: null);
 
-		String sequenceNumber = message.getHeaders().get(AwsHeaders.SEQUENCE_NUMBER, String.class);
+		String sequenceNumber = messageHeaders.get(AwsHeaders.SEQUENCE_NUMBER, String.class);
 		if (!StringUtils.hasText(sequenceNumber) && this.sequenceNumberExpression != null) {
 			sequenceNumber = this.sequenceNumberExpression.getValue(getEvaluationContext(), message, String.class);
 		}
 
 		Object payload = message.getPayload();
 
-		ByteBuffer data;
+		ByteBuffer data = null;
+
+		Message<?> messageToEmbed = null;
 
 		if (payload instanceof ByteBuffer) {
 			data = (ByteBuffer) payload;
+			if (this.embeddedHeadersMapper != null) {
+				messageToEmbed = new MutableMessage<>(data.array(), messageHeaders);
+			}
 		}
 		else {
 			byte[] bytes =
 					payload instanceof byte[]
 							? (byte[]) payload
 							: this.converter.convert(payload);
+			if (this.embeddedHeadersMapper != null) {
+				messageToEmbed = new MutableMessage<>(bytes, messageHeaders);
+			}
+			else {
+				data = ByteBuffer.wrap(bytes);
+			}
+		}
 
-			data = ByteBuffer.wrap(bytes);
+		if (messageToEmbed != null) {
+			data = ByteBuffer.wrap(this.embeddedHeadersMapper.fromMessage(messageToEmbed));
 		}
 
 		return new PutRecordRequest()

src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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,8 @@
 
 package org.springframework.integration.aws.outbound;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.Future;
 
 import org.springframework.cloud.aws.core.env.ResourceIdResolver;
@@ -25,6 +27,8 @@ import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.support.StandardTypeLocator;
 import org.springframework.integration.aws.support.AwsHeaders;
 import org.springframework.integration.aws.support.SnsBodyBuilder;
+import org.springframework.integration.aws.support.SnsHeaderMapper;
+import org.springframework.integration.mapping.HeaderMapper;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -32,6 +36,7 @@ import org.springframework.util.Assert;
 import com.amazonaws.AmazonWebServiceRequest;
 import com.amazonaws.handlers.AsyncHandler;
 import com.amazonaws.services.sns.AmazonSNSAsync;
+import com.amazonaws.services.sns.model.MessageAttributeValue;
 import com.amazonaws.services.sns.model.PublishRequest;
 import com.amazonaws.services.sns.model.PublishResult;
 
@@ -78,7 +83,7 @@ import com.amazonaws.services.sns.model.PublishResult;
  * @see PublishRequest
  * @see SnsBodyBuilder
  */
-public class SnsMessageHandler extends AbstractAwsMessageHandler {
+public class SnsMessageHandler extends AbstractAwsMessageHandler<Map<String, MessageAttributeValue>> {
 
 	private final AmazonSNSAsync amazonSns;
 
@@ -93,6 +98,7 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler {
 	public SnsMessageHandler(AmazonSNSAsync amazonSns) {
 		Assert.notNull(amazonSns, "amazonSns must not be null.");
 		this.amazonSns = amazonSns;
+		doSetHeaderMapper(new SnsHeaderMapper());
 	}
 
 	public void setTopicArn(String topicArn) {
@@ -185,6 +191,11 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler {
 			else {
 				publishRequest.setMessage(getConversionService().convert(snsMessage, String.class));
 			}
+
+			HeaderMapper<Map<String, MessageAttributeValue>> headerMapper = getHeaderMapper();
+			if (headerMapper != null) {
+				mapHeaders(message, publishRequest, headerMapper);
+			}
 		}
 
 		AsyncHandler<PublishRequest, PublishResult> asyncHandler = obtainAsyncHandler(message, publishRequest);
@@ -192,6 +203,16 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler {
 
 	}
 
+	private void mapHeaders(Message<?> message, PublishRequest publishRequest,
+			HeaderMapper<Map<String, MessageAttributeValue>> headerMapper) {
+
+		HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>();
+		headerMapper.fromHeaders(message.getHeaders(), messageAttributes);
+		if (!messageAttributes.isEmpty()) {
+			publishRequest.setMessageAttributes(messageAttributes);
+		}
+	}
+
 	@Override
 	protected void additionalOnSuccessHeaders(AbstractIntegrationMessageBuilder<?> messageBuilder,
 			AmazonWebServiceRequest request, Object result) {

src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.aws.outbound;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.Future;
 
 import org.springframework.cloud.aws.core.env.ResourceIdResolver;
@@ -23,8 +25,10 @@ import org.springframework.cloud.aws.messaging.support.destination.DynamicQueueU
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.aws.support.AwsHeaders;
+import org.springframework.integration.aws.support.SqsHeaderMapper;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.mapping.HeaderMapper;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.GenericMessageConverter;
@@ -36,6 +40,7 @@ import org.springframework.util.StringUtils;
 import com.amazonaws.AmazonWebServiceRequest;
 import com.amazonaws.handlers.AsyncHandler;
 import com.amazonaws.services.sqs.AmazonSQSAsync;
+import com.amazonaws.services.sqs.model.MessageAttributeValue;
 import com.amazonaws.services.sqs.model.SendMessageBatchRequest;
 import com.amazonaws.services.sqs.model.SendMessageBatchResult;
 import com.amazonaws.services.sqs.model.SendMessageRequest;
@@ -52,7 +57,7 @@ import com.amazonaws.services.sqs.model.SendMessageResult;
  * @see com.amazonaws.handlers.AsyncHandler
 
  */
-public class SqsMessageHandler extends AbstractAwsMessageHandler {
+public class SqsMessageHandler extends AbstractAwsMessageHandler<Map<String, MessageAttributeValue>> {
 
 	private final AmazonSQSAsync amazonSqs;
 
@@ -76,6 +81,7 @@ public class SqsMessageHandler extends AbstractAwsMessageHandler {
 	public SqsMessageHandler(AmazonSQSAsync amazonSqs, ResourceIdResolver resourceIdResolver) {
 		this.amazonSqs = amazonSqs;
 		this.destinationResolver = new DynamicQueueUrlDestinationResolver(amazonSqs, resourceIdResolver);
+		doSetHeaderMapper(new SqsHeaderMapper());
 	}
 
 	public void setQueue(String queue) {
@@ -187,12 +193,27 @@ public class SqsMessageHandler extends AbstractAwsMessageHandler {
 						this.messageDeduplicationIdExpression.getValue(getEvaluationContext(), message, String.class);
 				sendMessageRequest.setMessageDeduplicationId(messageDeduplicationId);
 			}
+
+			HeaderMapper<Map<String, MessageAttributeValue>> headerMapper = getHeaderMapper();
+			if (headerMapper != null) {
+				mapHeaders(message, sendMessageRequest, headerMapper);
+			}
 		}
 		AsyncHandler<SendMessageRequest, SendMessageResult> asyncHandler =
 				obtainAsyncHandler(message, sendMessageRequest);
 		return this.amazonSqs.sendMessageAsync(sendMessageRequest, asyncHandler);
 	}
 
+	private void mapHeaders(Message<?> message, SendMessageRequest sendMessageRequest,
+			HeaderMapper<Map<String, MessageAttributeValue>> headerMapper) {
+
+		HashMap<String, MessageAttributeValue> messageAttributes = new HashMap<>();
+		headerMapper.fromHeaders(message.getHeaders(), messageAttributes);
+		if (!messageAttributes.isEmpty()) {
+			sendMessageRequest.setMessageAttributes(messageAttributes);
+		}
+	}
+
 	@Override
 	protected void additionalOnSuccessHeaders(AbstractIntegrationMessageBuilder<?> messageBuilder,
 			AmazonWebServiceRequest request, Object result) {

src/main/java/org/springframework/integration/aws/support/AbstractMessageAttributesHeaderMapper.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2018 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.aws.support;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.aws.messaging.core.MessageAttributeDataTypes;
+import org.springframework.integration.mapping.HeaderMapper;
+import org.springframework.integration.util.PatternMatchUtils;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+import org.springframework.util.NumberUtils;
+
+/**
+ * Base {@link HeaderMapper} implementation for common logic in SQS and SNS
+ * around message attributes mapping.
+ *
+ * The {@link #toHeaders(Map)} is not supported.
+ *
+ * @param <A> the target message attribute type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 2.0
+ */
+public abstract class AbstractMessageAttributesHeaderMapper<A> implements HeaderMapper<Map<String, A>> {
+
+	private static final Log logger = LogFactory.getLog(SqsHeaderMapper.class);
+
+	private volatile String[] outboundHeaderNames = {
+			"!" + MessageHeaders.ID,
+			"!" + MessageHeaders.TIMESTAMP,
+			"!" + AwsHeaders.MESSAGE_ID,
+			"!" + AwsHeaders.QUEUE,
+			"!" + AwsHeaders.TOPIC,
+			"*" };
+
+	/**
+	 * Provide the header names that should be mapped to a AWS request object attributes
+	 * (for outbound adapters) from a Spring Integration Message's headers.
+	 * The values can also contain simple wildcard patterns (e.g. "foo*" or "*foo") to be matched.
+	 * Also supports negated ('!') patterns. First match wins (positive or negative).
+	 * To match the names starting with {@code !} symbol,
+	 * you have to escape it prepending with the {@code \} symbol in the pattern definition.
+	 * Defaults to map all ({@code *}) if the type is supported by SQS.
+	 * The {@link MessageHeaders#ID}, {@link MessageHeaders#TIMESTAMP},  {@link AwsHeaders#MESSAGE_ID},
+	 * {@link AwsHeaders#QUEUE} and {@link AwsHeaders#TOPIC} are ignored by default.
+	 * @param outboundHeaderNames The inbound header names.
+	 */
+	public void setOutboundHeaderNames(String... outboundHeaderNames) {
+		Assert.notNull(outboundHeaderNames, "'outboundHeaderNames' must not be null.");
+		Assert.noNullElements(outboundHeaderNames, "'outboundHeaderNames' must not contains null elements.");
+		Arrays.sort(outboundHeaderNames);
+		this.outboundHeaderNames = outboundHeaderNames;
+	}
+
+	@Override
+	public void fromHeaders(MessageHeaders headers, Map<String, A> target) {
+		for (Map.Entry<String, Object> messageHeader : headers.entrySet()) {
+			String messageHeaderName = messageHeader.getKey();
+			Object messageHeaderValue = messageHeader.getValue();
+
+			if (Boolean.TRUE.equals(PatternMatchUtils.smartMatch(messageHeaderName, this.outboundHeaderNames))) {
+
+				if (messageHeaderValue instanceof UUID
+						|| messageHeaderValue instanceof MimeType
+						|| messageHeaderValue instanceof Boolean
+						|| messageHeaderValue instanceof String) {
+
+					target.put(messageHeaderName, getStringMessageAttribute(messageHeaderValue.toString()));
+				}
+				else if (messageHeaderValue instanceof Number) {
+					target.put(messageHeaderName, getNumberMessageAttribute(messageHeaderValue));
+				}
+				else if (messageHeaderValue instanceof ByteBuffer) {
+					target.put(messageHeaderName, getBinaryMessageAttribute((ByteBuffer) messageHeaderValue));
+				}
+				else if (messageHeaderValue instanceof byte[]) {
+					target.put(messageHeaderName,
+							getBinaryMessageAttribute(ByteBuffer.wrap((byte[]) messageHeaderValue)));
+				}
+				else {
+					if (logger.isWarnEnabled()) {
+						logger.warn(
+								String.format("Message header with name '%s' and type '%s' cannot be sent as" +
+												" message attribute because it is not supported by SQS.",
+										messageHeaderName,
+										messageHeaderValue.getClass().getName()));
+					}
+				}
+
+			}
+		}
+	}
+
+	private A getBinaryMessageAttribute(ByteBuffer messageHeaderValue) {
+		return buildMessageAttribute(MessageAttributeDataTypes.BINARY, messageHeaderValue);
+	}
+
+	private A getStringMessageAttribute(String messageHeaderValue) {
+		return buildMessageAttribute(MessageAttributeDataTypes.STRING, messageHeaderValue);
+	}
+
+	private A getNumberMessageAttribute(Object messageHeaderValue) {
+		Assert.isTrue(NumberUtils.STANDARD_NUMBER_TYPES.contains(messageHeaderValue.getClass()),
+				"Only standard number types are accepted as message header.");
+
+		return buildMessageAttribute(MessageAttributeDataTypes.NUMBER + "." + messageHeaderValue.getClass().getName(),
+				messageHeaderValue);
+	}
+
+	protected abstract A buildMessageAttribute(String dataType, Object value);
+
+	@Override
+	public Map<String, Object> toHeaders(Map<String, A> source) {
+		throw new UnsupportedOperationException("The mapping from AWS Response Message is not supported");
+	}
+
+}

src/main/java/org/springframework/integration/aws/support/SnsHeaderMapper.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2018 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.aws.support;
+
+import java.nio.ByteBuffer;
+
+import com.amazonaws.services.sns.model.MessageAttributeValue;
+
+/**
+ * The {@link AbstractMessageAttributesHeaderMapper} implementation for the mapping
+ * from headers to SNS message attributes.
+ * <p>
+ * On the Inbound side, the SNS message is fully mapped from the JSON to the message payload.
+ * Only important HTTP headers are mapped to the message headers.
+ *
+ * @author Artem Bilan
+ *
+ * @since 2.0
+ */
+public class SnsHeaderMapper extends AbstractMessageAttributesHeaderMapper<MessageAttributeValue> {
+
+	@Override
+	protected MessageAttributeValue buildMessageAttribute(String dataType, Object value) {
+		MessageAttributeValue messageAttributeValue = new MessageAttributeValue()
+				.withDataType(dataType);
+		if (value instanceof ByteBuffer) {
+			return messageAttributeValue.withBinaryValue((ByteBuffer) value);
+		}
+		else {
+			return messageAttributeValue.withStringValue(value.toString());
+		}
+	}
+
+}

src/main/java/org/springframework/integration/aws/support/SqsHeaderMapper.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2018 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.aws.support;
+
+import java.nio.ByteBuffer;
+
+import org.springframework.messaging.MessageHeaders;
+
+import com.amazonaws.services.sqs.model.MessageAttributeValue;
+
+/**
+ * The {@link AbstractMessageAttributesHeaderMapper} implementation for the mapping
+ * from headers to SQS message attributes.
+ * <p>
+ * The {@link org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer} maps
+ * all the SQS message attributes to the {@link MessageHeaders}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 2.0
+ */
+public class SqsHeaderMapper extends AbstractMessageAttributesHeaderMapper<MessageAttributeValue> {
+
+	@Override
+	protected MessageAttributeValue buildMessageAttribute(String dataType, Object value) {
+		MessageAttributeValue messageAttributeValue = new MessageAttributeValue()
+				.withDataType(dataType);
+		if (value instanceof ByteBuffer) {
+			return messageAttributeValue.withBinaryValue((ByteBuffer) value);
+		}
+		else {
+			return messageAttributeValue.withStringValue(value.toString());
+		}
+	}
+
+}

src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.aws.kinesis;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
 
 import java.util.Date;
 import java.util.HashSet;
@@ -42,6 +43,7 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -95,11 +97,13 @@ public class KinesisIntegrationTests {
 		this.kinesisSendChannel.send(
 				MessageBuilder.withPayload(now)
 						.setHeader(AwsHeaders.STREAM, TEST_STREAM)
+						.setHeader("foo", "BAR")
 						.build());
 
 		Message<?> receive = this.kinesisReceiveChannel.receive(10_000);
 		assertThat(receive).isNotNull();
 		assertThat(receive.getPayload()).isEqualTo(now);
+		assertThat(receive.getHeaders()).contains(entry("foo", "BAR"));
 
 		Message<?> errorMessage = this.errorChannel.receive(10_000);
 		assertThat(errorMessage).isNotNull();
@@ -141,6 +145,7 @@ public class KinesisIntegrationTests {
 		public MessageHandler kinesisMessageHandler() {
 			KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(KINESIS_LOCAL_RUNNING.getKinesis());
 			kinesisMessageHandler.setPartitionKey("1");
+			kinesisMessageHandler.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo"));
 			return kinesisMessageHandler;
 		}
 
@@ -156,6 +161,7 @@ public class KinesisIntegrationTests {
 			adapter.setErrorChannel(errorChannel());
 			adapter.setErrorMessageStrategy(new KinesisMessageHeaderErrorMessageStrategy());
 			adapter.setCheckpointStore(checkpointStore());
+			adapter.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo"));
 			return adapter;
 		}
 

src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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,7 @@
 package org.springframework.integration.aws.outbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
@@ -38,6 +39,7 @@ import org.springframework.core.serializer.support.SerializingConverter;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.aws.support.AwsHeaders;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -77,7 +79,7 @@ public class KinesisMessageHandlerTests {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testKinesisMessageHandler() {
+	public void testKinesisMessageHandler() throws Exception {
 		Message<?> message = MessageBuilder.withPayload("message").build();
 		try {
 			this.kinesisSendChannel.send(message);
@@ -101,6 +103,7 @@ public class KinesisMessageHandlerTests {
 		message = MessageBuilder.fromMessage(message)
 				.setHeader(AwsHeaders.PARTITION_KEY, "fooKey")
 				.setHeader(AwsHeaders.SEQUENCE_NUMBER, "10")
+				.setHeader("foo", "bar")
 				.build();
 
 		this.kinesisSendChannel.send(message);
@@ -119,7 +122,12 @@ public class KinesisMessageHandlerTests {
 		assertThat(putRecordRequest.getPartitionKey()).isEqualTo("fooKey");
 		assertThat(putRecordRequest.getSequenceNumberForOrdering()).isEqualTo("10");
 		assertThat(putRecordRequest.getExplicitHashKey()).isNull();
-		assertThat(putRecordRequest.getData()).isEqualTo(ByteBuffer.wrap("message".getBytes()));
+
+		Message<?> messageToCheck = new EmbeddedJsonHeadersMessageMapper()
+				.toMessage(putRecordRequest.getData().array());
+
+		assertThat(messageToCheck.getHeaders()).contains(entry("foo", "bar"));
+		assertThat(messageToCheck.getPayload()).isEqualTo("message".getBytes());
 
 		AsyncHandler<?, ?> asyncHandler = asyncHandlerArgumentCaptor.getValue();
 
@@ -195,6 +203,7 @@ public class KinesisMessageHandlerTests {
 				}
 
 			});
+			kinesisMessageHandler.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo"));
 			return kinesisMessageHandler;
 		}
 

src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -22,6 +22,8 @@ import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 
+import java.util.Map;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
@@ -33,12 +35,14 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.aws.support.AwsHeaders;
 import org.springframework.integration.aws.support.SnsBodyBuilder;
+import org.springframework.integration.aws.support.SnsHeaderMapper;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.scheduling.annotation.AsyncResult;
 import org.springframework.test.annotation.DirtiesContext;
@@ -47,6 +51,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.amazonaws.handlers.AsyncHandler;
 import com.amazonaws.services.sns.AmazonSNSAsync;
+import com.amazonaws.services.sns.model.MessageAttributeValue;
 import com.amazonaws.services.sns.model.PublishRequest;
 import com.amazonaws.services.sns.model.PublishResult;
 
@@ -78,6 +83,7 @@ public class SnsMessageHandlerTests {
 		Message<?> message = MessageBuilder.withPayload(payload)
 				.setHeader("topic", "topic")
 				.setHeader("subject", "subject")
+				.setHeader("foo", "bar")
 				.build();
 
 		this.sendToSnsChannel.send(message);
@@ -96,6 +102,13 @@ public class SnsMessageHandlerTests {
 		assertThat(publishRequest.getMessage())
 				.isEqualTo("{\"default\":\"foo\",\"sms\":\"{\\\"foo\\\" : \\\"bar\\\"}\"}");
 
+		Map<String, MessageAttributeValue> messageAttributes = publishRequest.getMessageAttributes();
+
+		assertThat(messageAttributes).doesNotContainKey(MessageHeaders.ID);
+		assertThat(messageAttributes).doesNotContainKey(MessageHeaders.TIMESTAMP);
+		assertThat(messageAttributes).containsKey("foo");
+		assertThat(messageAttributes.get("foo").getStringValue()).isEqualTo("bar");
+
 		assertThat(reply.getHeaders().get(AwsHeaders.MESSAGE_ID)).isEqualTo("111");
 		assertThat(reply.getHeaders().get(AwsHeaders.TOPIC)).isEqualTo("topic");
 		assertThat(reply.getPayload()).isSameAs(payload);
@@ -135,6 +148,9 @@ public class SnsMessageHandlerTests {
 			snsMessageHandler.setSubjectExpression(PARSER.parseExpression("headers.subject"));
 			snsMessageHandler.setBodyExpression(PARSER.parseExpression("payload"));
 			snsMessageHandler.setOutputChannel(resultChannel());
+			SnsHeaderMapper headerMapper = new SnsHeaderMapper();
+			headerMapper.setOutboundHeaderNames("foo");
+			snsMessageHandler.setHeaderMapper(headerMapper);
 			return snsMessageHandler;
 		}
 

src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -23,6 +23,8 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import java.util.Map;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
@@ -39,6 +41,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -47,6 +50,7 @@ import com.amazonaws.handlers.AsyncHandler;
 import com.amazonaws.services.sqs.AmazonSQSAsync;
 import com.amazonaws.services.sqs.model.GetQueueUrlRequest;
 import com.amazonaws.services.sqs.model.GetQueueUrlResult;
+import com.amazonaws.services.sqs.model.MessageAttributeValue;
 import com.amazonaws.services.sqs.model.SendMessageRequest;
 
 /**
@@ -105,8 +109,18 @@ public class SqsMessageHandlerTests {
 		verify(this.amazonSqs, times(3))
 				.sendMessageAsync(sendMessageRequestArgumentCaptor.capture(), any(AsyncHandler.class));
 
-		assertThat(sendMessageRequestArgumentCaptor.getValue().getQueueUrl())
+		SendMessageRequest sendMessageRequestArgumentCaptorValue = sendMessageRequestArgumentCaptor.getValue();
+		assertThat(sendMessageRequestArgumentCaptorValue.getQueueUrl())
 				.isEqualTo("http://queue-url.com/baz");
+
+		Map<String, MessageAttributeValue> messageAttributes =
+				sendMessageRequestArgumentCaptorValue.getMessageAttributes();
+
+		assertThat(messageAttributes).doesNotContainKey(MessageHeaders.ID);
+		assertThat(messageAttributes).doesNotContainKey(MessageHeaders.TIMESTAMP);
+		assertThat(messageAttributes).containsKey("foo");
+		assertThat(messageAttributes.get("foo").getStringValue()).isEqualTo("baz");
+
 	}
 
 	@Configuration