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": "Refactor S3TransferManager to support non-file-based transfers"
+}

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)

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

@@ -19,11 +19,14 @@
 import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
+import java.util.UUID;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.sync.ResponseTransformer;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.testutils.RandomTempFile;
@@ -62,16 +65,17 @@ public static void teardown() throws IOException {
     }
 
     @Test
-    public void upload_fileSentCorrectly() throws IOException {
-        Upload upload = tm.upload(UploadRequest.builder()
-                                               .putObjectRequest(b -> b.bucket(TEST_BUCKET).key(TEST_KEY))
-                                               .source(testFile.toPath())
-                                               .overrideConfiguration(b -> b.addListener(LoggingTransferListener.create()))
-                                               .build());
+    public void upload_file_SentCorrectly() throws IOException {
+        FileUpload fileUpload =
+            tm.uploadFile(UploadFileRequest.builder()
+                                           .putObjectRequest(b -> b.bucket(TEST_BUCKET).key(TEST_KEY))
+                                           .source(testFile.toPath())
+                                           .overrideConfiguration(b -> b.addListener(LoggingTransferListener.create()))
+                                           .build());
 
-        CompletedUpload completedUpload = upload.completionFuture().join();
-        assertThat(completedUpload.response().responseMetadata().requestId()).isNotNull();
-        assertThat(completedUpload.response().sdkHttpResponse()).isNotNull();
+        CompletedFileUpload completedFileUpload = fileUpload.completionFuture().join();
+        assertThat(completedFileUpload.response().responseMetadata().requestId()).isNotNull();
+        assertThat(completedFileUpload.response().sdkHttpResponse()).isNotNull();
 
         ResponseInputStream<GetObjectResponse> obj = s3.getObject(r -> r.bucket(TEST_BUCKET).key(TEST_KEY),
                 ResponseTransformer.toInputStream());
@@ -80,4 +84,27 @@ public void upload_fileSentCorrectly() throws IOException {
                 .isEqualTo(ChecksumUtils.computeCheckSum(obj));
         assertThat(obj.response().responseMetadata().requestId()).isNotNull();
     }
+
+    @Test
+    public void upload_asyncRequestBody_SentCorrectly() throws IOException {
+        String content = UUID.randomUUID().toString();
+
+        Upload upload =
+            tm.upload(UploadRequest.builder()
+                                   .putObjectRequest(b -> b.bucket(TEST_BUCKET).key(TEST_KEY))
+                                   .requestBody(AsyncRequestBody.fromString(content))
+                                   .overrideConfiguration(b -> b.addListener(LoggingTransferListener.create()))
+                                   .build());
+
+        CompletedUpload completedUpload = upload.completionFuture().join();
+        assertThat(completedUpload.response().responseMetadata().requestId()).isNotNull();
+        assertThat(completedUpload.response().sdkHttpResponse()).isNotNull();
+
+        ResponseInputStream<GetObjectResponse> obj = s3.getObject(r -> r.bucket(TEST_BUCKET).key(TEST_KEY),
+                                                                  ResponseTransformer.toInputStream());
+
+        assertThat(ChecksumUtils.computeCheckSum(content.getBytes(StandardCharsets.UTF_8)))
+            .isEqualTo(ChecksumUtils.computeCheckSum(obj));
+        assertThat(obj.response().responseMetadata().requestId()).isNotNull();
+    }
 }

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/CompletedDirectoryTransfer.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.transfer.s3;
+
+import java.util.Collection;
+import software.amazon.awssdk.annotations.SdkPreviewApi;
+import software.amazon.awssdk.annotations.SdkPublicApi;
+
+/**
+ * A completed directory-based transfer.
+ * 
+ * @see CompletedDirectoryUpload
+ */
+@SdkPublicApi
+@SdkPreviewApi
+public interface CompletedDirectoryTransfer extends CompletedTransfer {
+
+    /**
+     * An immutable collection of failed transfers with error details, request metadata about each file that is failed to
+     * transfer.
+     *
+     * <p>
+     * Failed single object transfers can be retried by calling {@link S3TransferManager#uploadFile(UploadFileRequest)} or
+     * {@link S3TransferManager#downloadFile(DownloadFileRequest)}.
+     *
+     * <pre>
+     * {@code
+     * // Retrying failed uploads if the exception is retryable
+     * List<CompletableFuture<CompletedUpload>> futures =
+     *     completedDirectoryUpload.failedTransfers()
+     *                             .stream()
+     *                             .filter(failedSingleFileUpload -> isRetryable(failedSingleFileUpload.exception()))
+     *                             .map(failedSingleFileUpload ->
+     *                                  tm.upload(failedSingleFileUpload.request()).completionFuture())
+     *                             .collect(Collectors.toList());
+     * CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+     * }
+     * </pre>
+     *
+     * @return a list of failed transfers
+     */
+    Collection<? extends FailedObjectTransfer> failedTransfers();
+}