Refactor S3TransferManager to support non-file-based transfers (#2817)

* Refactor S3TransferManager to support non-file-based transfers

## Motivation

We would like for S3TransferManager to support non-file-based uploads
and downloads, such as reactive streams or in-memory byte buffers, but
the current implementation is directly coupled to file/Path-based
transfers. While the underlying S3AsyncClient already has support for
AsyncRequestBodies and AsyncResponseTransformers, these constructs are
not exposed through the current S3TransferManager interface.

Additionally, the existing S3TransferManager interfaces have made some
assumptions that make it difficult to introduce this support, such as
UploadRequest/DownloadRequest having a Path attribute, and not having
interfaces that allow us to distinguish between transfer types, e.g.,
single-object and multi-object (directory) transfers. Likewise, the
current interface hierarchy also makes it difficult to distinguish
between file-based and non-file-based. Therefore, this refactoring
requires backwards-incompatible changes, which are permissible while
S3TransferManager is still in PREVIEW.

## Modifications

The diff here is very large, but this change set primarily introduces
the following interface hierarchy:

* TransferRequest
    * TransferObjectRequest
        * UploadFileRequest
        * DownloadFileRequest
        * UploadRequest
        * DownloadRequest<T>
    * TransferDirectoryRequest
        * UploadDirectoryRequest
* Transfer
    * ObjectTransfer
        * FileUpload
        * FileDownload
        * Upload
        * Download<T>
    * DirectoryTransfer
        * DirectoryUpload
* CompletedTransfer
    * CompletedObjectTransfer
        * CompletedFileUpload
        * CompletedFileDownload
        * CompletedUpload
        * CompletedDownload<T>
    * CompletedDirectoryTransfer
        * CompletedDirectoryUpload
* FailedObjectTransfer
    * FailedFileUpload

These interfaces allow us to more selectively choose which attributes
will be shared between different data types. Additionally, we take this
opportunity to make the naming convention and class-vs-interface
distinction consistent across all the above data types.

It is important that we distinguish between file-based-requests and
non-file-based-requests so that the Collection<> of failed file
transfers from a directory upload can be re-driven, if needed.

Note that the above Download-oriented data types are all generic. This
is because of how AsyncResponseTransformer is designed to be generic,
being parameterized with the type of data that a given
AsyncResponseTransformer produces. Therefore, a user must declare the
type at the time of instantiating a DownloadRequest, and the type will
be consistent throughout the transfer lifecycle, including the Download
and CompletedDownload interfaces.

Author: Bennett-Lynch

.changes/next-release/feature-AmazonS3-2a00881.json

@@ -0,0 +1,6 @@
+{
+    "category": "Amazon S3", 
+    "contributor": "", 
+    "type": "feature", 
+    "description": "[Breaking Changes] Refactor S3TransferManager (PREVIEW) to support non-file-based transfers. This release refactors the S3TransferManager interface hierarchy and client API to differentiate between file-based and non-file-based transfers, allowing arbitrary object transfers. As a result, some S3TransferManager method signatures have changed in a backwards-incompatible way. Most notably, `Upload upload(UploadRequest)` becomes `FileUpload uploadFile(UploadFileRequest)`, and likewise for download variants. Please see https://github.com/aws/aws-sdk-java-v2/pull/2817 for a full list of changes."
+}

services-custom/s3-transfer-manager/README.md

@@ -19,47 +19,88 @@ First, you need to include the dependency in your project.
 ```
 
 Note that you need to replace `${awsjavasdk.version}` with the latest
-SDK version
+SDK version.
 
 ### Instantiate the transfer manager
-You can instantiate the transfer manager easily using the default settings
 
-```java
+You can instantiate the transfer manager easily using the default settings:
 
-S3TransferManager transferManager = S3TransferManager.create();
-    
+```java
+S3TransferManager tm = S3TransferManager.create();
 ```
 
 If you wish to configure settings, we recommend using the builder instead:
+
 ```java
-S3TransferManager transferManager = 
+S3TransferManager tm =
     S3TransferManager.builder()
                      .s3ClientConfiguration(b -> b.credentialsProvider(credentialProvider)
                                                   .region(Region.US_WEST_2)
                                                   .targetThroughputInGbps(20.0)
-                                                  .minimumPartSizeInBytes(10 * MB))
+                                                  .minimumPartSizeInBytes(8 * MB))
                      .build();
 ```
 
+### Upload a file to S3
+
+To upload a file to S3, you just need to provide the source file path and the `PutObjectRequest` that should be used for the upload:
+
+```java
+FileUpload upload =
+    tm.uploadFile(u -> u.source(Paths.get("myFile.txt"))
+                        .putObjectRequest(p -> p.bucket("bucket").key("key")));
+upload.completionFuture().join();
+```
+
 ### Download an S3 object to a file
-To download an object, you just need to provide the destination file path and the `GetObjectRequest` that should be used for the download.
+
+To download an object, you just need to provide the destination file path and the `GetObjectRequest` that should be used for the download:
 
 ```java
-Download download = 
-    transferManager.download(b -> b.destination(path)
-                                   .getObjectRequest(r -> r.bucket("bucket")
-                                                           .key("key")));
+FileDownload download =
+    tm.downloadFile(d -> d.getObjectRequest(g -> g.bucket("bucket").key("key"))
+                          .destination(Paths.get("myFile.txt")));
 download.completionFuture().join();
 ```
 
-### Upload a file to S3
-To upload a file to S3, you just need to provide the source file path and the `PutObjectRequest` that should be used for the upload.
+### Upload any content to S3
 
-```java
-Upload upload = transferManager.upload(b -> b.source(path)
-                                             .putObjectRequest(r -> r.bucket("bucket")
-                                                                     .key("key")));
+You may upload any arbitrary content to S3 by providing an `AsyncRequestBody`:
 
+```java
+Upload upload =
+    tm.upload(u -> u.requestBody(AsyncRequestBody.fromString("Hello world"))
+                    .putObjectRequest(p -> p.bucket("bucket").key("key")));
 upload.completionFuture().join();
+```
 
+Refer to the static factory methods available in `AsyncRequestBody` for other content sources.
+
+### Download an S3 object to a custom destination
+
+You may download an object from S3 to a custom destination by providing an `AsyncResponseTransformer`:
+
+*(This example buffers the entire object in memory and is not suitable for large objects.)*
+
+```java
+Download<ResponseBytes<GetObjectResponse>> download =
+    tm.download(d -> d.getObjectRequest(g -> g.bucket("bucket").key("key"))
+                      .responseTransformer(AsyncResponseTransformer.toBytes()));
+download.completionFuture().join();
+```
+
+Refer to the static factory methods available in `AsyncResponseTransformer` for other destinations.
+
+### Attach a TransferListener
+
+To monitor a transfer's progress, you can include a `TransferListener` with your transfer request:
+
+```java
+FileUpload upload =
+    tm.uploadFile(u -> u.source(Paths.get("myFile.txt"))
+                        .putObjectRequest(p -> p.bucket("bucket").key("key"))
+                        .overrideConfiguration(o -> o.addListener(LoggingTransferListener.create())));
+upload.completionFuture().join();
 ```
+
+You can provide your own implementation of a `TransferListener` to implement progress-bar-type functionality.

services-custom/s3-transfer-manager/src/it/java/software/amazon/awssdk/transfer/s3/CrtExceptionTransformationIntegrationTest.java

@@ -15,6 +15,12 @@
 
 package software.amazon.awssdk.transfer.s3;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -25,13 +31,6 @@
 import software.amazon.awssdk.testutils.RandomTempFile;
 import software.amazon.awssdk.transfer.s3.internal.S3CrtAsyncClient;
 
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
-
 public class CrtExceptionTransformationIntegrationTest extends S3IntegrationTestBase {
 
     private static final String BUCKET = temporaryBucketName(CrtExceptionTransformationIntegrationTest.class);
@@ -87,21 +86,21 @@ public void getObjectNoSuchBucket() throws IOException {
     @Test
     public void transferManagerDownloadObjectWithNoSuchKey() throws IOException {
         String randomBaseDirectory = Files.createTempDirectory(getClass().getSimpleName()).toString();
-        assertThatThrownBy(() -> transferManager.download(DownloadRequest.builder()
-                .getObjectRequest(GetObjectRequest.builder().bucket(BUCKET).key("randomKey").build())
-                .destination(Paths.get(randomBaseDirectory).resolve("testFile"))
-                .build()).completionFuture().join())
+        assertThatThrownBy(() -> transferManager.downloadFile(DownloadFileRequest.builder()
+                                                                                 .getObjectRequest(GetObjectRequest.builder().bucket(BUCKET).key("randomKey").build())
+                                                                                 .destination(Paths.get(randomBaseDirectory).resolve("testFile"))
+                                                                                 .build()).completionFuture().join())
                 .hasCauseInstanceOf(NoSuchKeyException.class)
                 .hasMessageContaining("software.amazon.awssdk.services.s3.model.NoSuchKeyException: The specified key does not exist");
     }
 
     @Test
     public void transferManagerDownloadObjectWithNoSuchBucket() throws IOException {
         String randomBaseDirectory = Files.createTempDirectory(getClass().getSimpleName()).toString();
-        assertThatThrownBy(() -> transferManager.download(DownloadRequest.builder()
-                .getObjectRequest(GetObjectRequest.builder().bucket("nonExistingTestBucket").key(KEY).build())
-                .destination(Paths.get(randomBaseDirectory).resolve("testFile"))
-                .build()).completionFuture().join())
+        assertThatThrownBy(() -> transferManager.downloadFile(DownloadFileRequest.builder()
+                                                                                 .getObjectRequest(GetObjectRequest.builder().bucket("nonExistingTestBucket").key(KEY).build())
+                                                                                 .destination(Paths.get(randomBaseDirectory).resolve("testFile"))
+                                                                                 .build()).completionFuture().join())
                 .hasCauseInstanceOf(NoSuchBucketException.class)
                 .hasMessageContaining("software.amazon.awssdk.services.s3.model.NoSuchBucketException: The specified bucket does not exist");
     }
@@ -127,10 +126,10 @@ public void putObjectNoSuchBucket() throws IOException {
 
     @Test
     public void transferManagerUploadObjectWithNoSuchObject() throws IOException{
-        assertThatThrownBy(() -> transferManager.upload(UploadRequest.builder()
-                .putObjectRequest(PutObjectRequest.builder().bucket("nonExistingTestBucket").key("someKey").build())
-                .source(testFile.toPath())
-                .build()).completionFuture().join())
+        assertThatThrownBy(() -> transferManager.uploadFile(UploadFileRequest.builder()
+                                                                             .putObjectRequest(PutObjectRequest.builder().bucket("nonExistingTestBucket").key("someKey").build())
+                                                                             .source(testFile.toPath())
+                                                                             .build()).completionFuture().join())
                 .hasCauseInstanceOf(NoSuchBucketException.class)
                 .hasMessageContaining("software.amazon.awssdk.services.s3.model.NoSuchBucketException: The specified bucket does not exist");
     }

services-custom/s3-transfer-manager/src/it/java/software/amazon/awssdk/transfer/s3/S3TransferManagerDownloadIntegrationTest.java

@@ -24,6 +24,9 @@
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.testutils.RandomTempFile;
 import software.amazon.awssdk.transfer.s3.progress.LoggingTransferListener;
@@ -46,7 +49,7 @@ public static void setup() throws IOException {
                                      .build(), file.toPath());
         tm = S3TransferManager.builder()
                               .s3ClientConfiguration(b -> b.region(DEFAULT_REGION)
-                                                                        .credentialsProvider(CREDENTIALS_PROVIDER_CHAIN))
+                                                           .credentialsProvider(CREDENTIALS_PROVIDER_CHAIN))
                               .build();
     }
 
@@ -58,15 +61,30 @@ public static void cleanup() {
     }
 
     @Test
-    public void download_shouldWork() throws IOException {
+    public void download_toFile() throws IOException {
         Path path = RandomTempFile.randomUncreatedFile().toPath();
-        Download download = tm.download(DownloadRequest.builder()
-                                                       .getObjectRequest(b -> b.bucket(BUCKET).key(KEY))
-                                                       .destination(path)
-                                                       .overrideConfiguration(b -> b.addListener(LoggingTransferListener.create()))
-                                                       .build());
-        CompletedDownload completedDownload = download.completionFuture().join();
+        FileDownload download =
+            tm.downloadFile(DownloadFileRequest.builder()
+                                               .getObjectRequest(b -> b.bucket(BUCKET).key(KEY))
+                                               .destination(path)
+                                               .overrideConfiguration(b -> b.addListener(LoggingTransferListener.create()))
+                                               .build());
+        CompletedFileDownload completedFileDownload = download.completionFuture().join();
         assertThat(Md5Utils.md5AsBase64(path.toFile())).isEqualTo(Md5Utils.md5AsBase64(file));
-        assertThat(completedDownload.response().responseMetadata().requestId()).isNotNull();
+        assertThat(completedFileDownload.response().responseMetadata().requestId()).isNotNull();
+    }
+
+    @Test
+    public void download_toBytes() throws Exception {
+        Download<ResponseBytes<GetObjectResponse>> download =
+            tm.download(DownloadRequest.builder()
+                                       .getObjectRequest(b -> b.bucket(BUCKET).key(KEY))
+                                       .responseTransformer(AsyncResponseTransformer.toBytes())
+                                       .overrideConfiguration(b -> b.addListener(LoggingTransferListener.create()))
+                                       .build());
+        CompletedDownload<ResponseBytes<GetObjectResponse>> completedDownload = download.completionFuture().join();
+        ResponseBytes<GetObjectResponse> result = completedDownload.result();
+        assertThat(Md5Utils.md5AsBase64(result.asByteArray())).isEqualTo(Md5Utils.md5AsBase64(file));
+        assertThat(result.response().responseMetadata().requestId()).isNotNull();
     }
 }

services-custom/s3-transfer-manager/src/it/java/software/amazon/awssdk/transfer/s3/S3TransferManagerUploadDirectoryIntegrationTest.java

@@ -85,12 +85,12 @@ public static void teardown() {
     @Test
     public void uploadDirectory_filesSentCorrectly() {
         String prefix = "yolo";
-        UploadDirectoryTransfer uploadDirectory = tm.uploadDirectory(u -> u.sourceDirectory(directory)
-                                                                           .bucket(TEST_BUCKET)
-                                                                           .prefix(prefix)
-                                                                           .overrideConfiguration(o -> o.recursive(true)));
-        CompletedUploadDirectory completedUploadDirectory = uploadDirectory.completionFuture().join();
-        assertThat(completedUploadDirectory.failedUploads()).isEmpty();
+        DirectoryUpload uploadDirectory = tm.uploadDirectory(u -> u.sourceDirectory(directory)
+                                                                          .bucket(TEST_BUCKET)
+                                                                          .prefix(prefix)
+                                                                          .overrideConfiguration(o -> o.recursive(true)));
+        CompletedDirectoryUpload completedDirectoryUpload = uploadDirectory.completionFuture().join();
+        assertThat(completedDirectoryUpload.failedTransfers()).isEmpty();
 
         List<String> keys =
             s3Client.listObjectsV2Paginator(b -> b.bucket(TEST_BUCKET).prefix(prefix)).contents().stream().map(S3Object::key)
@@ -105,13 +105,13 @@ public void uploadDirectory_filesSentCorrectly() {
     public void uploadDirectory_withDelimiter_filesSentCorrectly() {
         String prefix = "hello";
         String delimiter = "0";
-        UploadDirectoryTransfer uploadDirectory = tm.uploadDirectory(u -> u.sourceDirectory(directory)
-                                                                           .bucket(TEST_BUCKET)
-                                                                           .delimiter(delimiter)
-                                                                           .prefix(prefix)
-                                                                           .overrideConfiguration(o -> o.recursive(true)));
-        CompletedUploadDirectory completedUploadDirectory = uploadDirectory.completionFuture().join();
-        assertThat(completedUploadDirectory.failedUploads()).isEmpty();
+        DirectoryUpload uploadDirectory = tm.uploadDirectory(u -> u.sourceDirectory(directory)
+                                                                          .bucket(TEST_BUCKET)
+                                                                          .delimiter(delimiter)
+                                                                          .prefix(prefix)
+                                                                          .overrideConfiguration(o -> o.recursive(true)));
+        CompletedDirectoryUpload completedDirectoryUpload = uploadDirectory.completionFuture().join();
+        assertThat(completedDirectoryUpload.failedTransfers()).isEmpty();
 
         List<String> keys =
             s3Client.listObjectsV2Paginator(b -> b.bucket(TEST_BUCKET).prefix(prefix)).contents().stream().map(S3Object::key)