Add Spring Integration for S3 support (#1476)

* Copy all the S3 functionality from the Spring Integration AWS project
* Document the feature
* Add `spring-cloud-aws-starter-integration-s3` convenient module
* Introduce `LocalstackContainerTest` in the tests, similar to the one in DynamoDB module.
Reuse this abstraction for all the S3 integration tests to avoid unnecessary LocalStack container restarts
while we running all the tests in module

Author: artembilan

docs/src/main/asciidoc/s3.adoc

@@ -641,3 +641,149 @@ Sample IAM policy granting access to `spring-cloud-aws-demo` bucket:
     ]
 }
 ----
+
+=== 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 S3 Channel Adapters are based on the `S3Client` template and `S3TransferManager`.
+See their specification and Javadocs for more information.
+
+The S3 Inbound Channel Adapter is represented by the `S3InboundFileSynchronizingMessageSource` and allows pulling S3 objects as files from the S3 bucket to the local directory for synchronization.
+This adapter is fully similar to the Inbound Channel Adapters in the FTP and SFTP Spring Integration modules.
+See more information in the https://docs.spring.io/spring-integration/reference/ftp.html[FTP/FTPS Adapters Chapter] for common options or `SessionFactory`, `RemoteFileTemplate` and `FileListFilter` abstractions.
+
+The Java Configuration is:
+
+[source,java]
+----
+@SpringBootApplication
+public static class MyConfiguration {
+
+    @Autowired
+    private S3Client amazonS3;
+
+    @Bean
+    public S3InboundFileSynchronizer s3InboundFileSynchronizer() {
+    	S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(this.amazonS3);
+    	synchronizer.setDeleteRemoteFiles(true);
+    	synchronizer.setPreserveTimestamp(true);
+    	synchronizer.setRemoteDirectory(S3_BUCKET);
+    	synchronizer.setFilter(new S3RegexPatternFileListFilter(".*\\.test$"));
+    	Expression expression = PARSER.parseExpression("#this.toUpperCase() + '.a'");
+    	synchronizer.setLocalFilenameGeneratorExpression(expression);
+    	return synchronizer;
+    }
+
+    @Bean
+    @InboundChannelAdapter(value = "s3FilesChannel", poller = @Poller(fixedDelay = "100"))
+    public S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource() {
+    	S3InboundFileSynchronizingMessageSource messageSource =
+    			new S3InboundFileSynchronizingMessageSource(s3InboundFileSynchronizer());
+    	messageSource.setAutoCreateLocalDirectory(true);
+    	messageSource.setLocalDirectory(LOCAL_FOLDER);
+    	messageSource.setLocalFilter(new AcceptOnceFileListFilter<File>());
+    	return messageSource;
+    }
+
+    @Bean
+    public PollableChannel s3FilesChannel() {
+    	return new QueueChannel();
+    }
+}
+----
+
+With this config you receive messages with `java.io.File` `payload` from the `s3FilesChannel` after periodic synchronization of content from the Amazon S3 bucket into the local directory.
+
+The `S3StreamingMessageSource` adapter produces messages with payloads of type `InputStream`, allowing S3 objects to be fetched without writing to the local file system.
+Since the session remains open, the consuming application is responsible for closing the session when the file has been consumed.
+The session is provided in the closeableResource header (`IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE`).
+Standard framework components, such as the `FileSplitter` and `StreamTransformer` will automatically close the session.
+
+The following Spring Boot application provides an example of configuring the S3 inbound streaming adapter using Java configuration:
+
+[source,java]
+----
+@SpringBootApplication
+public class S3JavaApplication {
+
+    public static void main(String[] args) {
+        new SpringApplicationBuilder(S3JavaApplication.class)
+            .web(false)
+            .run(args);
+    }
+
+    @Autowired
+    private S3Client amazonS3;
+
+    @Bean
+    @InboundChannelAdapter(value = "s3Channel", poller = @Poller(fixedDelay = "100"))
+    public MessageSource<InputStream> s3InboundStreamingMessageSource() {
+        S3StreamingMessageSource messageSource = new S3StreamingMessageSource(template());
+        messageSource.setRemoteDirectory(S3_BUCKET);
+        messageSource.setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),
+                                   "streaming"));
+    	return messageSource;
+    }
+
+    @Bean
+    @Transformer(inputChannel = "s3Channel", outputChannel = "data")
+    public org.springframework.integration.transformer.Transformer transformer() {
+        return new StreamTransformer();
+    }
+
+    @Bean
+    public S3RemoteFileTemplate template() {
+        return new S3RemoteFileTemplate(new S3SessionFactory(this.amazonS3));
+    }
+
+    @Bean
+    public PollableChannel s3Channel() {
+    	return new QueueChannel();
+    }
+}
+----
+
+> NOTE: Unlike the non-streaming inbound channel adapter, this adapter does not prevent duplicates by default.
+> If you do not delete the remote file and wish to prevent the file being processed again, you can configure an `S3PersistentFileListFilter` in the `filter` attribute.
+> If you don’t want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter.
+> If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`.
+
+The `S3MessageHandler` is an Outbound Channel Adapter and allows performing `upload`, `download` and `copy` (see `S3MessageHandler.Command` enum) operations in the provided S3 bucket.
+
+The Java Configuration is:
+
+[source,java]
+----
+@SpringBootApplication
+public static class MyConfiguration {
+
+    @Autowired
+    private S3AsyncClient amazonS3;
+
+    @Bean
+    @ServiceActivator(inputChannel = "s3UploadChannel")
+    public MessageHandler s3MessageHandler() {
+    	return new S3MessageHandler(this.amazonS3, "my-bucket");
+    }
+
+}
+----
+
+With this config you can send a message with the `java.io.File` as `payload` and the `transferManager.upload()` operation will be performed, where the file name is used as a S3 Object key.
+
+See more information in the `S3MessageHandler` JavaDocs.
+
+NOTE: The AWS SDK recommends to use `S3CrtAsyncClient` for `S3TransferManager`, therefore an `S3AsyncClient.crtBuilder()` has to be used to achieve respective upload and download requirements, what is done internally in the `S3MessageHandler` when `S3CrtAsyncClient`-based constructor is used.
+
+The `S3MessageHandler` can be used as an Outbound Gateway with the `produceReply = true` constructor argument for Java Configuration.
+
+The "request-reply" nature of this gateway is async and the `Transfer` result from the `TransferManager` operation is sent to the `outputChannel`, assuming the transfer progress observation in the downstream flow.
+
+The `TransferListener` can be supplied to the `S3MessageHandler` to track the transfer progress per requests.
+
+See more information in the `S3MessageHandler` Javadocs.
+
+The Spring Integration dependency (as well as `s3-transfer-manager` and `aws-crt-client`) in the `spring-cloud-aws-s3` module are `optional` to avoid unnecessary artifacts on classpath when Spring Integration is not used.
+For convenience, a dedicated `spring-cloud-aws-starter-integration-s3` is provided managing all the required dependencies for Spring Integration support with Amazon S3.
+

pom.xml

@@ -52,6 +52,7 @@
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-metrics</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-parameter-store</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-s3</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-s3</module>
 		<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>

spring-cloud-aws-dependencies/pom.xml

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

spring-cloud-aws-s3/pom.xml

@@ -40,6 +40,11 @@
 			<artifactId>aws-crt-client</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-file</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>com.fasterxml.jackson.core</groupId>
 			<artifactId>jackson-databind</artifactId>
@@ -55,6 +60,11 @@
 			<artifactId>junit-jupiter</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 
 	</dependencies>
 </project>

spring-cloud-aws-s3/src/main/java/io/awspring/cloud/s3/integration/S3FileInfo.java

@@ -0,0 +1,88 @@
+/*
+ * 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.s3.integration;
+
+import java.util.Date;
+import org.springframework.integration.file.remote.AbstractFileInfo;
+import org.springframework.util.Assert;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+/**
+ * An Amazon S3 {@link org.springframework.integration.file.remote.FileInfo} implementation.
+ *
+ * @author Christian Tzolov
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class S3FileInfo extends AbstractFileInfo<S3Object> {
+
+	private final S3Object s3Object;
+
+	public S3FileInfo(S3Object s3Object) {
+		Assert.notNull(s3Object, "s3Object must not be null");
+		this.s3Object = s3Object;
+	}
+
+	@Override
+	public boolean isDirectory() {
+		return false;
+	}
+
+	@Override
+	public boolean isLink() {
+		return false;
+	}
+
+	@Override
+	public long getSize() {
+		return this.s3Object.size();
+	}
+
+	@Override
+	public long getModified() {
+		return this.s3Object.lastModified().getEpochSecond();
+	}
+
+	@Override
+	public String getFilename() {
+		return this.s3Object.key();
+	}
+
+	/**
+	 * A permissions representation string. Throws {@link UnsupportedOperationException} to avoid extra
+	 * {@link software.amazon.awssdk.services.s3.S3Client#getObjectAcl} REST call. The target application may choose to
+	 * do that according to its logic.
+	 * @return the permissions representation string.
+	 */
+	@Override
+	public String getPermissions() {
+		throw new UnsupportedOperationException("Use [AmazonS3.getObjectAcl()] to obtain permissions.");
+	}
+
+	@Override
+	public S3Object getFileInfo() {
+		return this.s3Object;
+	}
+
+	@Override
+	public String toString() {
+		return "FileInfo [isDirectory=" + isDirectory() + ", isLink=" + isLink() + ", Size=" + getSize()
+				+ ", ModifiedTime=" + new Date(getModified()) + ", Filename=" + getFilename() + ", RemoteDirectory="
+				+ getRemoteDirectory() + "]";
+	}
+
+}

spring-cloud-aws-s3/src/main/java/io/awspring/cloud/s3/integration/S3InboundFileSynchronizer.java

@@ -0,0 +1,87 @@
+/*
+ * 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.s3.integration;
+
+import java.io.File;
+import java.io.IOException;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.file.remote.session.Session;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+import org.springframework.lang.Nullable;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+/**
+ * An implementation of {@link AbstractInboundFileSynchronizer} for Amazon S3.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class S3InboundFileSynchronizer extends AbstractInboundFileSynchronizer<S3Object> {
+
+	public S3InboundFileSynchronizer() {
+		this(new S3SessionFactory());
+	}
+
+	public S3InboundFileSynchronizer(S3Client amazonS3) {
+		this(new S3SessionFactory(amazonS3));
+	}
+
+	/**
+	 * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances.
+	 * @param sessionFactory The session factory.
+	 */
+	@SuppressWarnings("this-escape")
+	public S3InboundFileSynchronizer(SessionFactory<S3Object> sessionFactory) {
+		super(sessionFactory);
+		doSetRemoteDirectoryExpression(new LiteralExpression(null));
+		doSetFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3MessageSource"));
+	}
+
+	@Override
+	protected boolean isFile(S3Object file) {
+		return true;
+	}
+
+	@Override
+	protected String getFilename(S3Object file) {
+		return (file != null ? file.key() : null);
+	}
+
+	@Override
+	protected long getModified(S3Object file) {
+		return file.lastModified().getEpochSecond();
+	}
+
+	@Override
+	protected boolean copyFileToLocalDirectory(String remoteDirectoryPath,
+			@Nullable EvaluationContext localFileEvaluationContext, S3Object remoteFile, File localDirectory,
+			Session<S3Object> session) throws IOException {
+
+		return super.copyFileToLocalDirectory(((S3Session) session).normalizeBucketName(remoteDirectoryPath),
+				localFileEvaluationContext, remoteFile, localDirectory, session);
+	}
+
+	@Override
+	protected String protocol() {
+		return "s3";
+	}
+
+}