Refactors

Author: alextwoods

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

@@ -19,7 +19,6 @@
 import static software.amazon.awssdk.services.s3.crt.S3CrtSdkHttpExecutionAttribute.CRT_PROGRESS_LISTENER;
 import static software.amazon.awssdk.services.s3.crt.S3CrtSdkHttpExecutionAttribute.METAREQUEST_PAUSE_OBSERVABLE;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.CRT_PAUSE_RESUME_TOKEN;
-import static software.amazon.awssdk.services.s3.multipart.S3MultipartExecutionAttribute.JAVA_PROGRESS_LISTENER;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
@@ -70,36 +69,14 @@ public final Upload upload(UploadRequest uploadRequest) {
         TransferProgressUpdater progressUpdater = new TransferProgressUpdater(uploadRequest,
                                                                               requestBody.contentLength().orElse(null));
         progressUpdater.transferInitiated();
-        // requestBody = progressUpdater.wrapRequestBody(requestBody);
         progressUpdater.registerCompletion(returnFuture);
 
-        S3MetaRequestPauseObservable observable = new S3MetaRequestPauseObservable();
-
-        Consumer<SdkHttpExecutionAttributes.Builder> attachObservable =
-            b -> b.put(METAREQUEST_PAUSE_OBSERVABLE, observable)
-                  .put(CRT_PROGRESS_LISTENER, progressUpdater.crtProgressListener());
-
-        PutObjectRequest putObjectRequest = attachCrtSdkAttribute(uploadRequest.putObjectRequest(), attachObservable);
+        Consumer<SdkHttpExecutionAttributes.Builder> attachProgress =
+            b -> b.put(CRT_PROGRESS_LISTENER, progressUpdater.crtProgressListener());
 
-        progressUpdater.transferInitiated();
-        progressUpdater.registerCompletion(returnFuture);
+        PutObjectRequest putObjectRequest = attachCrtSdkAttribute(uploadRequest.putObjectRequest(), attachProgress);
 
-        try {
-            assertNotUnsupportedArn(uploadRequest.putObjectRequest().bucket(), "upload");
-
-            CompletableFuture<PutObjectResponse> future =
-                s3AsyncClient.putObject(putObjectRequest, requestBody);
-
-            // Forward upload cancellation to future
-            CompletableFutureUtils.forwardExceptionTo(returnFuture, future);
-
-            CompletableFutureUtils.forwardTransformedResultTo(future, returnFuture,
-                                                              r -> CompletedUpload.builder()
-                                                                                  .response(r)
-                                                                                  .build());
-        } catch (Throwable throwable) {
-            returnFuture.completeExceptionally(throwable);
-        }
+        doUpload(putObjectRequest, requestBody, returnFuture);
 
         return new DefaultUpload(returnFuture, progressUpdater.progress());
     }

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

@@ -148,24 +148,31 @@ public Upload upload(UploadRequest uploadRequest) {
             putObjectRequest = attachSdkAttribute(uploadRequest.putObjectRequest(), attachProgressListener);
         }
 
+        doUpload(putObjectRequest, requestBody, returnFuture);
+
+        return new DefaultUpload(returnFuture, progressUpdater.progress());
+    }
+
+    protected void doUpload(PutObjectRequest putObjectRequest, AsyncRequestBody requestBody,
+                    CompletableFuture<CompletedUpload> returnFuture) {
         try {
-            assertNotUnsupportedArn(uploadRequest.putObjectRequest().bucket(), "upload");
+            assertNotUnsupportedArn(putObjectRequest.bucket(), "upload");
 
             CompletableFuture<PutObjectResponse> future =
                 s3AsyncClient.putObject(putObjectRequest, requestBody);
 
             // Forward upload cancellation to future
             CompletableFutureUtils.forwardExceptionTo(returnFuture, future);
 
-            CompletableFutureUtils.forwardTransformedResultTo(future, returnFuture,
-                                                              r -> CompletedUpload.builder()
-                                                                                  .response(r)
-                                                                                  .build());
+            CompletableFutureUtils.forwardTransformedResultTo(
+                future,
+                returnFuture,
+                r -> CompletedUpload.builder()
+                                    .response(r)
+                                    .build());
         } catch (Throwable throwable) {
             returnFuture.completeExceptionally(throwable);
         }
-
-        return new DefaultUpload(returnFuture, progressUpdater.progress());
     }
 
     /**

services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/CrtS3TransferManagerTest.java

@@ -20,6 +20,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute.SDK_HTTP_EXECUTION_ATTRIBUTES;
+import static software.amazon.awssdk.services.s3.crt.S3CrtSdkHttpExecutionAttribute.CRT_PROGRESS_LISTENER;
 import static software.amazon.awssdk.services.s3.crt.S3CrtSdkHttpExecutionAttribute.METAREQUEST_PAUSE_OBSERVABLE;
 
 import com.google.common.jimfs.Jimfs;
@@ -105,6 +106,6 @@ private void verifyCrtInRequestAttributes() {
         assertThat(actual.overrideConfiguration()).isPresent();
         SdkHttpExecutionAttributes attribute = actual.overrideConfiguration().get().executionAttributes().getAttribute(SDK_HTTP_EXECUTION_ATTRIBUTES);
         assertThat(attribute).isNotNull();
-        assertThat(attribute.getAttribute(METAREQUEST_PAUSE_OBSERVABLE)).isNotNull();
+        assertThat(attribute.getAttribute(CRT_PROGRESS_LISTENER)).isNotNull();
     }
 }