diff --git a/README.md b/README.md index ec4fe68..ed9e1ec 100644 --- a/README.md +++ b/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 diff --git a/build.gradle b/build.gradle index c21051f..f940376 100644 --- a/build.gradle +++ b/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' diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index ce03dd6..dea1d2e 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/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 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 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 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)) { diff --git a/src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java index 7eb54ef..e0c6410 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/AbstractAwsMessageHandler.java +++ b/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 the headers container type. + * * @author Artem Bilan * * @since 2.0 */ -public abstract class AbstractAwsMessageHandler extends AbstractMessageProducingHandler { +public abstract class AbstractAwsMessageHandler 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 headerMapper; + public void setAsyncHandler(AsyncHandler 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 headerMapper) { + doSetHeaderMapper(headerMapper); + } + + protected final void doSetHeaderMapper(HeaderMapper headerMapper) { + this.headerMapper = headerMapper; + } + + protected HeaderMapper 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); diff --git a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java index 26bc77c..0b297b5 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java +++ b/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 { private final AmazonKinesisAsync amazonKinesis; private Converter 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 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 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 headerMapper) { + throw new UnsupportedOperationException("Kinesis doesn't support headers.\n" + + "Consider to use 'OutboundMessageMapper' for embedding headers into the record data."); + } + + @Override + protected Future handleMessageToAws(Message message) throws Exception { if (message.getPayload() instanceof PutRecordsRequest) { AsyncHandler 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() diff --git a/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java index e8d2955..72c95ae 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/SnsMessageHandler.java +++ b/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> { 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> headerMapper = getHeaderMapper(); + if (headerMapper != null) { + mapHeaders(message, publishRequest, headerMapper); + } } AsyncHandler asyncHandler = obtainAsyncHandler(message, publishRequest); @@ -192,6 +203,16 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler { } + private void mapHeaders(Message message, PublishRequest publishRequest, + HeaderMapper> headerMapper) { + + HashMap messageAttributes = new HashMap<>(); + headerMapper.fromHeaders(message.getHeaders(), messageAttributes); + if (!messageAttributes.isEmpty()) { + publishRequest.setMessageAttributes(messageAttributes); + } + } + @Override protected void additionalOnSuccessHeaders(AbstractIntegrationMessageBuilder messageBuilder, AmazonWebServiceRequest request, Object result) { diff --git a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java index 445864b..1a62399 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/SqsMessageHandler.java +++ b/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> { 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> headerMapper = getHeaderMapper(); + if (headerMapper != null) { + mapHeaders(message, sendMessageRequest, headerMapper); + } } AsyncHandler asyncHandler = obtainAsyncHandler(message, sendMessageRequest); return this.amazonSqs.sendMessageAsync(sendMessageRequest, asyncHandler); } + private void mapHeaders(Message message, SendMessageRequest sendMessageRequest, + HeaderMapper> headerMapper) { + + HashMap messageAttributes = new HashMap<>(); + headerMapper.fromHeaders(message.getHeaders(), messageAttributes); + if (!messageAttributes.isEmpty()) { + sendMessageRequest.setMessageAttributes(messageAttributes); + } + } + @Override protected void additionalOnSuccessHeaders(AbstractIntegrationMessageBuilder messageBuilder, AmazonWebServiceRequest request, Object result) { diff --git a/src/main/java/org/springframework/integration/aws/support/AbstractMessageAttributesHeaderMapper.java b/src/main/java/org/springframework/integration/aws/support/AbstractMessageAttributesHeaderMapper.java new file mode 100644 index 0000000..a8df5af --- /dev/null +++ b/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 the target message attribute type. + * + * @author Artem Bilan + * + * @since 2.0 + */ +public abstract class AbstractMessageAttributesHeaderMapper implements HeaderMapper> { + + 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 target) { + for (Map.Entry 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 toHeaders(Map source) { + throw new UnsupportedOperationException("The mapping from AWS Response Message is not supported"); + } + +} diff --git a/src/main/java/org/springframework/integration/aws/support/SnsHeaderMapper.java b/src/main/java/org/springframework/integration/aws/support/SnsHeaderMapper.java new file mode 100644 index 0000000..7dc7445 --- /dev/null +++ b/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. + *

+ * 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 { + + @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()); + } + } + +} diff --git a/src/main/java/org/springframework/integration/aws/support/SqsHeaderMapper.java b/src/main/java/org/springframework/integration/aws/support/SqsHeaderMapper.java new file mode 100644 index 0000000..919334d --- /dev/null +++ b/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. + *

+ * 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 { + + @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()); + } + } + +} diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java index 89a894a..8545d8b 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/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; } diff --git a/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java index 2833405..654effc 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/KinesisMessageHandlerTests.java +++ b/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; } diff --git a/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java index 486fa3c..536bfdf 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/SnsMessageHandlerTests.java +++ b/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 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; } diff --git a/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java index 889d198..f5432af 100644 --- a/src/test/java/org/springframework/integration/aws/outbound/SqsMessageHandlerTests.java +++ b/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 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