Add Spring Integration channel adapter for SQS

docs/src/main/asciidoc/sqs.adoc

@@ -2235,3 +2235,46 @@ Sample IAM policy granting access to SQS:
             "Resource": "yourARN"
         }
 ----
+
+=== Spring Integration Support
+
+Starting with version 4.0, Spring Cloud AWS provides https://spring.io/projects/spring-integration[Spring Integration] channel adapters for Amazon SQS.
+
+The `SqsMessageHandler` is for publishing a single message (or their batch) to SQS queue configured explicitly on the `SqsMessageHandler` or resolved via SpEL expression against the request message.
+The logic of this `MessageHandler` is to consume Spring Integration messages from an `inputChannel` and internally it is heavily base on the `SqsAsyncOperations` mentioned above.
+When the `SqsMessageHandler` is set into an `async` mode, the result of the send operation is produced as a reply message into the `outputChannel`.
+For a single request, the reply message is created based on the `SendResult`.
+With request message payload as a `Collection<Message<?>>`, the `SqsAsyncOperations.sendManyAsync()` is performed; the `SendResult.Batch` is produced as is as a payload of the reply message.
+The minimal configuration for this channel adapter is as follows:
+
+[source,java]
+----
+@Bean
+@ServiceActivator(inputChannel = "sqsSendChannel")
+MessageHandler sqsMessageHandler(SqsAsyncOperations sqsAsyncOperations) {
+    SqsMessageHandler sqsMessageHandler = new SqsMessageHandler(sqsAsyncOperations);
+    sqsMessageHandler.setQueue("queue1");
+    return sqsMessageHandler;
+}
+----
+
+The `SqsMessageDrivenChannelAdapter` is for consuming messages from one or more SQS queues.
+This channel adapter requires an `SqsAsyncClient` and internally is heavily based on the `SqsMessageListenerContainer` mentioned above.
+The `SqsContainerOptions` can be injected for further listener container customization.
+The consumed messages are then produced to the `outputChannel` for further Spring Integration processing.
+If the `ListenerMode` on the `SqsContainerOptions` is set to `BATCH`, the received `Collection<Message<?>>` is wrapped into a single message to produce.
+The Spring Integration https://docs.spring.io/spring-integration/reference/splitter.html[splitter] pattern could be used downstream for per-message processing.
+The minimal configuration for this channel adapter is as follows:
+
+[source,java]
+----
+@Bean
+MessageProducer sqsMessageDrivenChannelAdapter(SqsAsyncClient sqsAsyncClient) {
+    SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(sqsAsyncClient, "testQueue");
+    adapter.setOutputChannelName("inputChannel");
+    return adapter;
+}
+----
+
+The Spring Integration dependency in the `spring-cloud-aws-sqs` module is `optinal` to avoid unnecessary artifacts on classpath when Spring Integration is not used.
+For convenience, a dedicated `spring-cloud-aws-starter-integration-sqs` is provided managing all the required dependencies for Spring Integration support with Amazon SQS.

spring-cloud-aws-dependencies/pom.xml

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

spring-cloud-aws-sqs/pom.xml

@@ -30,6 +30,11 @@
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-context</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.retry</groupId>
 			<artifactId>spring-retry</artifactId>

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/integration/SqsMessageDrivenChannelAdapter.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2025-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.sqs.integration;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import io.awspring.cloud.sqs.config.SqsMessageListenerContainerFactory;
+import io.awspring.cloud.sqs.listener.MessageListener;
+import io.awspring.cloud.sqs.listener.SqsContainerOptions;
+import io.awspring.cloud.sqs.listener.SqsMessageListenerContainer;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.support.management.IntegrationManagedResource;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link MessageProducerSupport} implementation for the Amazon SQS
+ * {@code receiveMessage}. Works in 'listener' manner and delegates hard work to the
+ * {@link SqsMessageListenerContainer}.
+ *
+ * @author Artem Bilan
+ * @author Patrick Fitzsimons
+ *
+ * @since 4.0
+ *
+ * @see SqsMessageListenerContainerFactory
+ * @see MessageListener
+ */
+@ManagedResource
+@IntegrationManagedResource
+public class SqsMessageDrivenChannelAdapter extends MessageProducerSupport {
+
+	private final SqsMessageListenerContainerFactory.Builder<Object> sqsMessageListenerContainerFactory =
+		SqsMessageListenerContainerFactory.builder();
+
+	private final String[] queues;
+
+	private SqsContainerOptions sqsContainerOptions;
+
+	private SqsMessageListenerContainer<?> listenerContainer;
+
+	public SqsMessageDrivenChannelAdapter(SqsAsyncClient amazonSqs, String... queues) {
+		Assert.noNullElements(queues, "'queues' must not be empty");
+		this.sqsMessageListenerContainerFactory.sqsAsyncClient(amazonSqs);
+		this.queues = Arrays.copyOf(queues, queues.length);
+	}
+
+	public void setSqsContainerOptions(SqsContainerOptions sqsContainerOptions) {
+		this.sqsContainerOptions = sqsContainerOptions;
+	}
+
+	@Override
+	protected void onInit() {
+		super.onInit();
+		if (this.sqsContainerOptions != null) {
+			this.sqsMessageListenerContainerFactory.configure(sqsContainerOptionsBuilder ->
+				sqsContainerOptionsBuilder.fromBuilder(this.sqsContainerOptions.toBuilder()));
+		}
+		this.sqsMessageListenerContainerFactory.messageListener(new IntegrationMessageListener());
+		this.listenerContainer = this.sqsMessageListenerContainerFactory.build().createContainer(this.queues);
+	}
+
+	@Override
+	public String getComponentType() {
+		return "aws:sqs-message-driven-channel-adapter";
+	}
+
+	@Override
+	protected void doStart() {
+		this.listenerContainer.start();
+	}
+
+	@Override
+	protected void doStop() {
+		this.listenerContainer.stop();
+	}
+
+	@ManagedAttribute
+	public String[] getQueues() {
+		return Arrays.copyOf(this.queues, this.queues.length);
+	}
+
+	private class IntegrationMessageListener implements MessageListener<Object> {
+
+		IntegrationMessageListener() {
+		}
+
+		@Override
+		public void onMessage(Message<Object> message) {
+			sendMessage(message);
+		}
+
+		@Override
+		public void onMessage(Collection<Message<Object>> messages) {
+			onMessage(new GenericMessage<>(messages));
+		}
+
+	}
+
+}

spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/integration/SqsMessageHandler.java

@@ -0,0 +1,171 @@
+/*
+ * Copyright 2025-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.sqs.integration;
+
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import io.awspring.cloud.sqs.listener.SqsHeaders;
+import io.awspring.cloud.sqs.operations.SendResult;
+import io.awspring.cloud.sqs.operations.SqsAsyncOperations;
+
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.MessageTimeoutException;
+import org.springframework.integration.expression.ExpressionUtils;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link AbstractMessageProducingHandler} implementation for the Amazon SQS.
+ * All the logic based on the {@link SqsAsyncOperations#sendAsync(String, Message)}
+ * or {@link SqsAsyncOperations#sendManyAsync(String, Collection)} if the request message's payload
+ * is a collection of {@link Message} instances.
+ * <p>
+ * All the SQS-specific message attributes have to be provided in the respective message headers
+ * via {@link SqsHeaders.MessageSystemAttributes} constant values or with the {@link SqsAsyncOperations}.
+ * <p>
+ * This {@link AbstractMessageProducingHandler} produces a reply only in the {@link #isAsync()} mode.
+ * For a single request message the {@link SendResult} is converted to the reply message with respective headers.
+ * The {@link SendResult.Batch} is sent as a reply message's payload as is.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ *
+ * @see SqsAsyncOperations#sendAsync
+ * @see SqsAsyncOperations#sendManyAsync
+ * @see SqsHeaders.MessageSystemAttributes
+ */
+public class SqsMessageHandler extends AbstractMessageProducingHandler {
+
+	public static final long DEFAULT_SEND_TIMEOUT = 10000;
+
+	private final SqsAsyncOperations sqsAsyncOperations;
+
+	private Expression queueExpression;
+
+	private EvaluationContext evaluationContext;
+
+	private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT);
+
+	public SqsMessageHandler(SqsAsyncOperations sqsAsyncOperations) {
+		this.sqsAsyncOperations = sqsAsyncOperations;
+	}
+
+	public void setQueue(String queue) {
+		setQueueExpression(new LiteralExpression(queue));
+	}
+
+	public void setQueueExpressionString(String queueExpression) {
+		setQueueExpression(EXPRESSION_PARSER.parseExpression(queueExpression));
+	}
+
+	public void setQueueExpression(Expression queueExpression) {
+		this.queueExpression = queueExpression;
+	}
+
+	public void setSendTimeout(long sendTimeout) {
+		setSendTimeoutExpression(new ValueExpression<>(sendTimeout));
+	}
+
+	public void setSendTimeoutExpressionString(String sendTimeoutExpression) {
+		setSendTimeoutExpression(EXPRESSION_PARSER.parseExpression(sendTimeoutExpression));
+	}
+
+	public void setSendTimeoutExpression(Expression sendTimeoutExpression) {
+		Assert.notNull(sendTimeoutExpression, "'sendTimeoutExpression' must not be null");
+		this.sendTimeoutExpression = sendTimeoutExpression;
+	}
+
+	@Override
+	protected void onInit() {
+		Assert.notNull(this.queueExpression, "The SQS queue must be provided.");
+		super.onInit();
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
+	}
+
+	@Override
+	protected boolean shouldCopyRequestHeaders() {
+		return false;
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	protected void handleMessageInternal(Message<?> message) {
+		String queueName = this.queueExpression.getValue(this.evaluationContext, message, String.class);
+		Assert.hasText(queueName, "The 'queueExpression' must not evaluate to empty String.");
+		CompletableFuture<?> resultFuture;
+		if (message.getPayload() instanceof Collection<?> collection) {
+			Assert.notEmpty(collection, "The payload with a collection of messages must not be empty.");
+			Object next = collection.iterator().next();
+			Assert.isInstanceOf(Message.class, next,
+				"The payload with a collection of messages must contain 'Message' instances only.");
+			Collection<Message<Object>> messages = (Collection<Message<Object>>) collection;
+
+			resultFuture = this.sqsAsyncOperations.sendManyAsync(queueName, messages)
+				.thenApply((batchResult) -> getMessageBuilderFactory().withPayload(batchResult).build());
+		}
+		else {
+			resultFuture = this.sqsAsyncOperations.sendAsync(queueName, message)
+				.thenApply((sendResult) ->
+					getMessageBuilderFactory()
+						.fromMessage(sendResult.message())
+						.setHeader(SqsHeaders.SQS_QUEUE_NAME_HEADER, sendResult.endpoint())
+						.setHeader(SqsHeaders.MessageSystemAttributes.MESSAGE_ID, sendResult.messageId())
+						.copyHeaders(sendResult.additionalInformation())
+						.build());
+		}
+
+		if (isAsync()) {
+			sendOutputs(resultFuture, message);
+			return;
+		}
+
+		Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class);
+		if (sendTimeout == null || sendTimeout < 0) {
+			try {
+				resultFuture.get();
+			}
+			catch (InterruptedException | ExecutionException ex) {
+				throw new IllegalStateException(ex);
+			}
+		}
+		else {
+			try {
+				resultFuture.get(sendTimeout, TimeUnit.MILLISECONDS);
+			}
+			catch (TimeoutException te) {
+				throw new MessageTimeoutException(message, "Timeout waiting for response from Amazon SQS", te);
+			}
+			catch (InterruptedException ex) {
+				Thread.currentThread().interrupt();
+				throw new IllegalStateException(ex);
+			}
+			catch (ExecutionException ex) {
+				throw new IllegalStateException(ex);
+			}
+		}
+	}
+
+}