Add Spring Integration channel adapters for SNS (#1473)

* Add Spring Integration channel adapters for SNS

* Add `spring-integration-http` as an optional dep into `spring-cloud-aws-sns`.
* Add `SnsInboundChannelAdapter` to consume SNS notification over HTTP(S)
* Add `SnsMessageHandler` to publish to an SNS topic.
* The `SnsHeaderMapper` and `SnsBodyBuilder` are supporting classes for various SNS publish scenarios
* The `SnsRequestFailureException` is a generic exception with a message a and request context
to throw when publication fails
* Add extra useful `SnsHeaders` constant for some Spring Integration message headers
* Since `SnsMessageHandler` is based on the `SnsAsyncClient`, introduce an `SnsAsyncTopicArnResolver`
for such an async use-case
* Add `spring-cloud-aws-starter-integration-sns`
* Document this new feature
* Add missed `spring-cloud-aws-starter-integration-sqs` module int the root pom

* * Fix typos in docs and Javadocs for SI SNS
* Simplify `try..catch` logic in the `SnsMessageHandler` and add missed `currentThread().interrupt()`
* Add `.fifo` logic to the `SnsAsyncTopicArnResolver`

Author: artembilan

docs/src/main/asciidoc/sns.adoc

@@ -296,3 +296,93 @@ Sample IAM policy granting access to SNS:
     ]
 }
 ----
+
+=== Spring Integration Support
+
+Starting with version 4.0, Spring Cloud AWS provides https://spring.io/projects/spring-integration[Spring Integration] channel adapters for Amazon SNS.
+
+The `SnsInboundChannelAdapter` is an extension of `HttpRequestHandlingMessagingGateway` and must be as a part of Spring MVC application.
+Its URL must be used from the AWS Management Console to add this endpoint as a subscriber to the SNS Topic.
+However, before receiving any notification itself, this HTTP endpoint must confirm the subscription.
+
+See `SnsInboundChannelAdapter` JavaDocs for more information.
+
+An important option of this adapter to consider is `handleNotificationStatus`.
+This `boolean` flag indicates if the adapter should send `SubscriptionConfirmation/UnsubscribeConfirmation` message to the `output-channel` or not.
+If that is a case, the `SnsHeaders.NOTIFICATION_STATUS_HEADER` message header is present in the message with the `NotificationStatus` object, which can be used in the downstream flow to confirm subscription or not.
+Or "re-confirm" it in the case of `UnsubscribeConfirmation` message.
+
+In addition, the `SnsHeaders.SNS_MESSAGE_TYPE_HEADER` message header is represented to simplify a routing in the downstream flow.
+
+The Java Configuration is pretty simple:
+
+[source,java]
+----
+@SpringBootApplication
+public static class MyConfiguration {
+
+	@Autowired
+	private SnsClient amazonSns;
+
+	@Bean
+    public PollableChannel inputChannel() {
+    	return new QueueChannel();
+    }
+
+    @Bean
+    public HttpRequestHandler sqsMessageDrivenChannelAdapter(PollableChannel inputChannel) {
+    	SnsInboundChannelAdapter adapter = new SnsInboundChannelAdapter(this.amazonSns, "/mySampleTopic");
+    	adapter.setRequestChannel(inputChannel);
+    	adapter.setHandleNotificationStatus(true);
+    	return adapter;
+    }
+}
+----
+
+Note: by default, the message `payload` is a `Map` converted from the received Topic JSON message.
+For the convenience a `payload-expression` is provided with the `Message` as a root object of the evaluation context.
+Hence, even some HTTP headers, populated by the `DefaultHttpHeaderMapper`, are available for the evaluation context.
+
+
+The `SnsMessageHandler` is a simple one-way Outbound Channel Adapter to send Topic Notification using `SnsAsyncClient` service.
+
+This Channel Adapter (`MessageHandler`) accepts these options:
+
+- `topic-arn` (`topic-arn-expression`) - the SNS Topic to send notification for.
+- `subject` (`subject-expression`) - the SNS Notification Subject;
+- `body-expression` - the SpEL expression to evaluate the `message` property for the `software.amazon.awssdk.services.sns.model.PublishRequest`.
+- `resource-id-resolver` - a `ResourceIdResolver` bean reference to resolve logical topic names to physical resource ids;
+
+See `SnsMessageHandler` JavaDocs for more information.
+
+The Java Config looks like:
+
+[source,java]
+----
+@Bean
+public MessageHandler snsMessageHandler(SnsAsyncClient amazonSns) {
+    SnsMessageHandler handler = new SnsMessageHandler(amazonSns);
+    handler.setTopicArn("arn:aws:sns:eu-west:123456789012:test");
+    String bodyExpression = "T(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(new FunctionExpression<Message<?>>(m -> m.getHeaders().get(MessageHeaders.ID)));
+    return handler;
+}
+----
+
+NOTE: the `bodyExpression` can be evaluated to a `io.awspring.cloud.sns.integration.SnsBodyBuilder` allowing the configuration of a `json` `messageStructure` for the `PublishRequest` and provide separate messages for different protocols.
+The same `SnsBodyBuilder` rule is applied for the raw `payload` if the `bodyExpression` hasn't been configured.
+
+NOTE: if the `payload` of `requestMessage` is a `software.amazon.awssdk.services.sns.model.PublishRequest` already, the `SnsMessageHandler` doesn't do anything with it, and it is sent as-is.
+
+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 https://docs.aws.amazon.com/sns/latest/dg/SNSMessageAttributes.html[Amazon SNS Message Attributes] about value types and restrictions.
+
+The `SnsMessageHandler` supports sending to SNS FIFO topics using the `messageGroupId`/`messageGroupIdExpression` and `messageDeduplicationIdExpression` properties.
+
+The Spring Integration dependency in the `spring-cloud-aws-sns` module is `optional` to avoid unnecessary artifacts on classpath when Spring Integration is not used.
+For convenience, a dedicated `spring-cloud-aws-starter-integration-sns` is provided managing all the required dependencies for Spring Integration support with Amazon SNS.

pom.xml

@@ -55,7 +55,9 @@
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-secrets-manager</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-ses</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-sns</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-sns</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-sqs</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-sqs</module>
 		<module>spring-cloud-aws-samples</module>
 		<module>spring-cloud-aws-test</module>
 		<module>spring-cloud-aws-modulith</module>

spring-cloud-aws-dependencies/pom.xml

@@ -208,6 +208,12 @@
 				<version>${project.version}</version>
 			</dependency>
 
+			<dependency>
+				<groupId>io.awspring.cloud</groupId>
+				<artifactId>spring-cloud-aws-starter-integration-sns</artifactId>
+				<version>${project.version}</version>
+			</dependency>
+
 			<dependency>
 				<groupId>io.awspring.cloud</groupId>
 				<artifactId>spring-cloud-aws-starter-sqs</artifactId>

spring-cloud-aws-sns/pom.xml

@@ -36,6 +36,11 @@
 			<artifactId>spring-web</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-http</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-webmvc</artifactId>

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/SnsAsyncTopicArnResolver.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2013-2025 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
+ *
+ *      https://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 io.awspring.cloud.sns.core;
+
+import java.util.Map;
+import org.springframework.util.Assert;
+import software.amazon.awssdk.arns.Arn;
+import software.amazon.awssdk.services.sns.SnsAsyncClient;
+import software.amazon.awssdk.services.sns.model.CreateTopicRequest;
+
+/**
+ * A {@link TopicArnResolver} implementation to determine topic ARN by name against an {@link SnsAsyncClient}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class SnsAsyncTopicArnResolver implements TopicArnResolver {
+
+	private final SnsAsyncClient snsClient;
+
+	public SnsAsyncTopicArnResolver(SnsAsyncClient snsClient) {
+		Assert.notNull(snsClient, "snsClient is required");
+		this.snsClient = snsClient;
+	}
+
+	/**
+	 * Resolve topic ARN by topic name. If topicName is already an ARN, it returns {@link Arn}. If topicName is just a
+	 * string with a topic name, it attempts to create a topic, or if the topic already exists, just returns its ARN.
+	 */
+	@Override
+	public Arn resolveTopicArn(String topicName) {
+		Assert.notNull(topicName, "topicName must not be null");
+		if (topicName.toLowerCase().startsWith("arn:")) {
+			return Arn.fromString(topicName);
+		}
+		else {
+			CreateTopicRequest.Builder builder = CreateTopicRequest.builder().name(topicName);
+
+			// fix for https://github.com/awspring/spring-cloud-aws/issues/707
+			if (topicName.endsWith(".fifo")) {
+				builder.attributes(Map.of("FifoTopic", "true"));
+			}
+
+			// if the topic exists, createTopic returns a successful response with the topic arn
+			return Arn.fromString(this.snsClient.createTopic(builder.build()).join().topicArn());
+		}
+	}
+
+}

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/SnsHeaders.java

@@ -15,17 +15,26 @@
  */
 package io.awspring.cloud.sns.core;
 
+import io.awspring.cloud.sns.handlers.NotificationStatus;
+import io.awspring.cloud.sns.integration.SnsInboundChannelAdapter;
 import org.springframework.messaging.Message;
 import software.amazon.awssdk.services.sns.model.PublishRequest;
+import software.amazon.awssdk.services.sns.model.PublishResponse;
 
 /**
  * SNS specific headers that can be applied to Spring Messaging {@link Message}.
  *
  * @author Matej Nedic
+ * @author Artem Bilan
  * @since 3.0.0
  */
 public final class SnsHeaders {
 
+	/**
+	 * SNS Headers prefix to be used by all headers added by the framework.
+	 */
+	public static final String SNS_HEADER_PREFIX = "Sns_";
+
 	/**
 	 * Notification subject. The value of this header is set to {@link PublishRequest#subject()}.
 	 */
@@ -43,6 +52,29 @@ public final class SnsHeaders {
 	 */
 	public static final String MESSAGE_DEDUPLICATION_ID_HEADER = "message-deduplication-id";
 
+	/**
+	 * Topic ARN header where the message has been published. The value of this header is set from
+	 * {@link PublishRequest#topicArn()}.
+	 */
+	public static final String TOPIC_HEADER = SNS_HEADER_PREFIX + "topicArn";
+
+	/**
+	 * Message id header as a unique identifier assigned to the published message, or from a received message. The value
+	 * of this header is set from {@link PublishResponse#messageId()}.
+	 */
+	public static final String MESSAGE_ID_HEADER = SNS_HEADER_PREFIX + "messageId";
+
+	/**
+	 * The {@link NotificationStatus} header for manual confirmation on reception. The value of this header is set from
+	 * {@link SnsInboundChannelAdapter}.
+	 */
+	public static final String NOTIFICATION_STATUS_HEADER = SNS_HEADER_PREFIX + "notificationStatus";
+
+	/**
+	 * The {@value SNS_MESSAGE_TYPE_HEADER} header for the received SNS message type.
+	 */
+	public static final String SNS_MESSAGE_TYPE_HEADER = SNS_HEADER_PREFIX + "messageType";
+
 	private SnsHeaders() {
 
 	}

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/integration/SnsBodyBuilder.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2013-2025 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
+ *
+ *      https://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 io.awspring.cloud.sns.integration;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.springframework.util.Assert;
+
+/**
+ * A utility class to simplify an SNS Message body building. Can be used from the
+ * {@code SnsMessageHandler#bodyExpression} definition or directly in case of manual
+ * {@link software.amazon.awssdk.services.sns.model.PublishRequest} building.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public final class SnsBodyBuilder {
+
+	private final Map<String, String> snsMessage = new HashMap<>();
+
+	private SnsBodyBuilder(String defaultMessage) {
+		Assert.hasText(defaultMessage, "defaultMessage must not be empty.");
+		this.snsMessage.put("default", defaultMessage);
+	}
+
+	public SnsBodyBuilder forProtocols(String message, String... protocols) {
+		Assert.hasText(message, "message must not be empty.");
+		Assert.notEmpty(protocols, "protocols must not be empty.");
+		for (String protocol : protocols) {
+			Assert.hasText(protocol, "protocols must not contain empty elements.");
+			this.snsMessage.put(protocol, message);
+		}
+		return this;
+	}
+
+	public String build() {
+		StringBuilder stringBuilder = new StringBuilder("{");
+		for (Map.Entry<String, String> entry : this.snsMessage.entrySet()) {
+			stringBuilder.append("\"").append(entry.getKey()).append("\":\"")
+					.append(entry.getValue().replaceAll("\"", "\\\\\"")).append("\",");
+		}
+		return stringBuilder.substring(0, stringBuilder.length() - 1) + "}";
+	}
+
+	public static SnsBodyBuilder withDefault(String defaultMessage) {
+		return new SnsBodyBuilder(defaultMessage);
+	}
+
+}