GH-196: Add SNS FIFO support

Fixes #196

* Add support for SNS FIFO message group and deduplication IDs
* Add Javadoc, clean up formatting
* Fix Javadoc formatting
* Add SNS FIFO information to README
* Change XML snippet indentation to tabs
* Code samples throughout the README mix tabs and spaces, but this
snippet was using tabs, so this updates the new lines to match.
* Minor code style clean up

**Cherry-pick to `main`**

Author: chrylis

README.md

@@ -437,9 +437,14 @@ The Java Config looks like:
 @Bean
 public MessageHandler snsMessageHandler() {
     SnsMessageHandler handler = new SnsMessageHandler(amazonSns());
-    adapter.setTopicArn("arn:aws:sns:eu-west:123456789012:test);
+    adapter.setTopicArn("arn:aws:sns:eu-west:123456789012:test");
     String bodyExpression = "SnsBodyBuilder.withDefault(payload).forProtocols(payload.substring(0, 140), 'sms')";
     handler.setBodyExpression(spelExpressionParser.parseExpression(bodyExpression));
+
+    // message-group ID and deduplication ID are used for FIFO topics
+    handler.setMessageGroupId("foo-messages");
+    String deduplicationExpression = "headers.id";
+    handler.setMessageDeduplicationIdExpression(spelExpressionParser.parseExpression(deduplicationExpression))''
     return handler;
 }
 ````
@@ -457,12 +462,17 @@ The XML variant may look like:
 			channel="notificationChannel"
 			topic-arn="foo"
 			subject="bar"
+			message-group-id="foo-messages"
+			message-deduplication-id-expression="headers.id"
 			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.   
 
+Starting with _version 2.5.3_, the `SnsMessageHandler` supports sending to SNS FIFO topics using the `messageGroupId`/`messageGroupIdExpression`
+and `messageDeduplicationIdExpression` properties.
+
 ## 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. 

src/main/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -30,6 +30,7 @@
  * The parser for the {@code <int-aws:sns-outbound-channel-adapter>}.
  *
  * @author Artem Bilan
+ * @author Christopher Smith
  */
 public class SnsOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
 
@@ -42,6 +43,8 @@ protected AbstractBeanDefinition parseConsumer(Element element, ParserContext pa
 
 		AwsParserUtils.populateExpressionAttribute("topic-arn", builder, element, parserContext);
 		AwsParserUtils.populateExpressionAttribute("subject", builder, element, parserContext);
+		AwsParserUtils.populateExpressionAttribute("message-group-id", builder, element, parserContext);
+		AwsParserUtils.populateExpressionAttribute("message-deduplication-id", builder, element, parserContext);
 
 		BeanDefinition message = IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("body-expression",
 				element);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 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 java.util.Map;
 import java.util.concurrent.Future;
 
+import org.springframework.core.log.LogMessage;
 import org.springframework.expression.Expression;
 import org.springframework.expression.TypeLocator;
 import org.springframework.expression.common.LiteralExpression;
@@ -71,6 +72,8 @@
  * </ul>
  *
  * @author Artem Bilan
+ * @author Christopher Smith
+ *
  * @see AmazonSNSAsync
  * @see PublishRequest
  * @see SnsBodyBuilder
@@ -83,6 +86,10 @@ public class SnsMessageHandler extends AbstractAwsMessageHandler<Map<String, Mes
 
 	private Expression subjectExpression;
 
+	private Expression messageGroupIdExpression;
+
+	private Expression messageDeduplicationIdExpression;
+
 	private Expression bodyExpression;
 
 	private ResourceIdResolver resourceIdResolver;
@@ -113,10 +120,51 @@ public void setSubjectExpression(Expression subjectExpression) {
 		this.subjectExpression = subjectExpression;
 	}
 
+	/**
+	 * A fixed message-group ID to be set for messages sent to an SNS FIFO topic
+	 * from this handler.
+	 * Equivalent to calling {{@link #setMessageGroupIdExpression(Expression)} with
+	 * a literal string expression.
+	 * @param messageGroupId the group ID to be used for all messages sent from this handler
+	 * @since 2.5.3
+	 */
+	public void setMessageGroupId(String messageGroupId) {
+		Assert.hasText(messageGroupId, "messageGroupId must not be empty.");
+		this.messageGroupIdExpression = new LiteralExpression(messageGroupId);
+	}
+
+
+	/**
+	 * The {@link Expression} to determine the
+	 * <a href="https://docs.aws.amazon.com/sns/latest/dg/fifo-message-grouping.html">message group</a>
+	 * for messages sent to an SNS FIFO topic from this handler.
+	 * @param messageGroupIdExpression the {@link Expression} to produce the message-group ID
+	 * @since 2.5.3
+	 */
+	public void setMessageGroupIdExpression(Expression messageGroupIdExpression) {
+		Assert.notNull(messageGroupIdExpression, "messageGroupIdExpression must not be null.");
+		this.messageGroupIdExpression = messageGroupIdExpression;
+	}
+
+	/**
+	 * The {@link Expression} to determine the deduplication ID for this message.
+	 * SNS FIFO topics
+	 * <a href="https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html">require a message deduplication ID to be specified</a>,
+	 * either in the adapter configuration or on a {@link PublishRequest} payload
+	 * of the request {@link Message}, unless content-based deduplication is enabled
+	 * on the topic.
+	 * @param messageDeduplicationIdExpression the {@link Expression} to produce the message deduplication ID
+	 * @since 2.5.3
+	 */
+	public void setMessageDeduplicationIdExpression(Expression messageDeduplicationIdExpression) {
+		Assert.notNull(messageDeduplicationIdExpression, "messageDeduplicationIdExpression must not be null.");
+		this.messageDeduplicationIdExpression = messageDeduplicationIdExpression;
+	}
+
 	/**
 	 * The {@link Expression} to produce the SNS notification message. If it evaluates to
 	 * the {@link SnsBodyBuilder} the {@code messageStructure} of the
-	 * {@link PublishRequest} is set to {@code json}. Otherwise the
+	 * {@link PublishRequest} is set to {@code json}. Otherwise, the
 	 * {@link #getConversionService()} is used to convert the evaluation result to the
 	 * {@link String} without setting the {@code messageStructure}.
 	 * @param bodyExpression the {@link Expression} to produce the SNS notification
@@ -172,6 +220,25 @@ protected Future<?> handleMessageToAws(Message<?> message) {
 				publishRequest.setSubject(subject);
 			}
 
+			if (this.messageGroupIdExpression != null) {
+				if (!topicArn.endsWith(".fifo")) {
+					logger.warn(LogMessage.format("a messageGroupId will be set for non-FIFO topic '%s'", topicArn));
+				}
+				String messageGroupId =
+						this.messageGroupIdExpression.getValue(getEvaluationContext(), message, String.class);
+				publishRequest.setMessageGroupId(messageGroupId);
+			}
+
+			if (this.messageDeduplicationIdExpression != null) {
+				if (!topicArn.endsWith(".fifo")) {
+					logger.warn(
+							LogMessage.format("a messageDeduplicationId will be set for non-FIFO topic '%s'", topicArn));
+				}
+				String messageDeduplicationId =
+						this.messageDeduplicationIdExpression.getValue(getEvaluationContext(), message, String.class);
+				publishRequest.setMessageDeduplicationId(messageDeduplicationId);
+			}
+
 			Object snsMessage = message.getPayload();
 
 			if (this.bodyExpression != null) {

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

@@ -531,7 +531,7 @@
 					Asynchronous callback handler for events in the lifecycle of the request. Users can provide an
 					implementation of the callback methods in this interface to receive notification of successful or
 					unsuccessful completion of the operation.
-					By default successful reply is sent to the 'success-channel' and error message to the
+					By default, successful reply is sent to the 'success-channel' and error message to the
 					'failure-channel' if they are provided.
 				</xsd:documentation>
 			</xsd:annotation>
@@ -960,7 +960,7 @@
 					A SpEL expression that resolves to an Amazon SNS Topic ARN.
 					The 'requestMessage' is the root object for evaluation context.
 					Mutually exclusive with 'topic-arn'.
-					This attribute isn't mandatory and the the topic can be specified on the
+					This attribute isn't mandatory and the topic can be specified on the
 					'com.amazonaws.services.sns.model.PublishRequest'
 					payload of the request Message.
 				</xsd:documentation>
@@ -985,6 +985,39 @@
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="message-group-id">
+			<xsd:annotation>
+				<xsd:documentation>
+					The message group ID.
+					Mutually exclusive with 'message-group-id-expression'.
+					SNS FIFO topics require a message group to be specified, either in
+					the adapter configuration or on a 'PublishRequest' payload
+					of the request Message.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="message-group-id-expression">
+			<xsd:annotation>
+				<xsd:documentation>
+					The SpEL expression for the message group ID.
+					Mutually exclusive with 'message-group-id'.
+					SNS FIFO topics require a message group to be specified, either in
+					the adapter configuration or on a 'PublishRequest' payload
+					of the request Message.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="message-deduplication-id-expression">
+			<xsd:annotation>
+				<xsd:documentation>
+					The SpEL expression for the message deduplication ID.
+					SNS FIFO topics require a message deduplication ID to be specified, either in
+					the adapter configuration or on a 'PublishRequest' payload
+					of the request Message, unless content-based deduplication is enabled
+					on the topic.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="body-expression">
 			<xsd:annotation>
 				<xsd:documentation>

src/test/java/org/springframework/integration/aws/config/xml/SnsOutboundChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -36,6 +36,7 @@
 
 /**
  * @author Artem Bilan
+ * @author Christopher Smith
  */
 @SpringJUnitConfig
 @DirtiesContext
@@ -79,6 +80,8 @@ void testSnsOutboundChannelAdapterDefaultParser() {
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "amazonSns")).isSameAs(this.amazonSns);
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "evaluationContext")).isNotNull();
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "topicArnExpression")).isNull();
+		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "messageGroupIdExpression")).isNull();
+		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "messageDeduplicationIdExpression")).isNull();
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "subjectExpression")).isNull();
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "bodyExpression")).isNull();
 		assertThat(TestUtils.getPropertyValue(this.defaultAdapterHandler, "resourceIdResolver"))

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -55,6 +55,7 @@
 
 /**
  * @author Artem Bilan
+ * @author Christopher Smith
  */
 @SpringJUnitConfig
 @DirtiesContext
@@ -92,6 +93,8 @@ void testSnsMessageHandler() {
 		assertThat(publishRequest.getMessageStructure()).isEqualTo("json");
 		assertThat(publishRequest.getTopicArn()).isEqualTo("topic");
 		assertThat(publishRequest.getSubject()).isEqualTo("subject");
+		assertThat(publishRequest.getMessageGroupId()).isEqualTo("SUBJECT");
+		assertThat(publishRequest.getMessageDeduplicationId()).isEqualTo("BAR");
 		assertThat(publishRequest.getMessage())
 				.isEqualTo("{\"default\":\"foo\",\"sms\":\"{\\\"foo\\\" : \\\"bar\\\"}\"}");
 
@@ -136,6 +139,8 @@ public PollableChannel resultChannel() {
 		public MessageHandler snsMessageHandler() {
 			SnsMessageHandler snsMessageHandler = new SnsMessageHandler(amazonSNS());
 			snsMessageHandler.setTopicArnExpression(PARSER.parseExpression("headers.topic"));
+			snsMessageHandler.setMessageGroupIdExpression(PARSER.parseExpression("headers.subject.toUpperCase()"));
+			snsMessageHandler.setMessageDeduplicationIdExpression(PARSER.parseExpression("headers.foo.toUpperCase()"));
 			snsMessageHandler.setSubjectExpression(PARSER.parseExpression("headers.subject"));
 			snsMessageHandler.setBodyExpression(PARSER.parseExpression("payload"));
 			snsMessageHandler.setOutputChannel(resultChannel());