Added integ test

Author: alextwoods

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

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.transfer.s3;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
 
 import java.io.IOException;
@@ -215,4 +216,31 @@ void upload_file_Interupted_CancelsTheListener(S3TransferManager tm) {
         assertThat(transferListener.getRatioTransferredList().get(transferListener.getRatioTransferredList().size() - 1))
             .isNotEqualTo(100.0);
     }
+
+    @ParameterizedTest
+    @MethodSource("transferManagers")
+    void upload_asyncRequestBody_ReportsProgressCorrectly(S3TransferManager tm) throws IOException {
+        String content = RandomStringUtils.randomAscii(OBJ_SIZE);
+        CaptureTransferListener transferListener = new CaptureTransferListener();
+
+        Upload upload =
+            tm.upload(UploadRequest.builder()
+                                   .putObjectRequest(b -> b.bucket(TEST_BUCKET).key(TEST_KEY))
+                                   .requestBody(AsyncRequestBody.fromString(content))
+                                   .addTransferListener(LoggingTransferListener.create())
+                                   .addTransferListener(transferListener)
+                                   .build());
+
+        upload.completionFuture().join();
+        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));
+
+        assertListenerForSuccessfulTransferComplete(transferListener);
+
+        // ensure intermediate progress is reported
+        assertThat(transferListener.getRatioTransferredList()).hasSizeGreaterThan(2);
+    }
 }