Java v2: Update S3 examples of uploading from a stream of unknown size to remove use of BlockingInputStreamAsyncRequestBody

.doc_gen/metadata/s3_metadata.yaml

@@ -3294,13 +3294,15 @@ s3_Scenario_UploadStream:
             - description: Use the <ulink url="sdk-for-java/latest/developer-guide/crt-based-s3-client.html" type="documentation">&AWS;
                 CRT-based S3 Client</ulink>.
               snippet_tags:
-                - s3.java2.async_stream.import
-                - s3.java2.async_stream.main
+                - s3.java2.async_stream.complete
+            - description: Use the standard <ulink url="sdk-for-java/latest/developer-guide/s3-async-client-multipart.html#s3-async-client-mp-on" type="documentation">
+                asynchronous S3 client with multipart upload enabled</ulink>.
+              snippet_tags:
+                - s3.java2.async_stream_mp.complete
             - description: Use the <ulink url="sdk-for-java/latest/developer-guide/transfer-manager.html" type="documentation">&S3;
                 Transfer Manager</ulink>.
               snippet_tags:
-                - s3.tm.java2.upload_stream.import
-                - s3.tm.java2.upload_stream.main
+                - s3.tm.java2.upload_stream.complete
     Swift:
       versions:
         - sdk_version: 1

javav2/example_code/s3/src/main/java/com/example/s3/async/PutObjectFromStreamAsync.java

@@ -4,18 +4,21 @@
 
 // snippet-start:[s3.java2.async_stream.complete]
 // snippet-start:[s3.java2.async_stream.import]
+
 import com.example.s3.util.AsyncExampleUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
-import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 
 import java.io.ByteArrayInputStream;
+import java.io.InputStream;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 // snippet-end:[s3.java2.async_stream.import]
 
 public class PutObjectFromStreamAsync {
@@ -28,7 +31,8 @@ public static void main(String[] args) {
         AsyncExampleUtils.createBucket(bucketName);
         try {
             PutObjectFromStreamAsync example = new PutObjectFromStreamAsync();
-            PutObjectResponse putObjectResponse = example.putObjectFromStream(AsyncExampleUtils.client, bucketName, key);
+            S3AsyncClient s3AsyncClientCrt = S3AsyncClient.crtCreate();
+            PutObjectResponse putObjectResponse = example.putObjectFromStreamCrt(s3AsyncClientCrt, bucketName, key);
             logger.info("Object {} etag: {}", key, putObjectResponse.eTag());
             logger.info("Object {} uploaded to bucket {}.", key, bucketName);
         } catch (SdkException e) {
@@ -41,29 +45,29 @@ public static void main(String[] args) {
 
 // snippet-start:[s3.java2.async_stream.main]
     /**
-     * @param s33CrtAsyncClient - To upload content from a stream of unknown size, use the AWS CRT-based S3 client. For more information, see
-     *                          https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/crt-based-s3-client.html.
+     * @param s33CrtAsyncClient - To upload content from a stream of unknown size, use can the AWS CRT-based S3 client.
      * @param bucketName - The name of the bucket.
      * @param key - The name of the object.
      * @return software.amazon.awssdk.services.s3.model.PutObjectResponse - Returns metadata pertaining to the put object operation.
      */
-    public PutObjectResponse putObjectFromStream(S3AsyncClient s33CrtAsyncClient, String bucketName, String key) {
-
-        BlockingInputStreamAsyncRequestBody body =
-                AsyncRequestBody.forBlockingInputStream(null); // 'null' indicates a stream will be provided later.
-
-        CompletableFuture<PutObjectResponse> responseFuture =
-                s33CrtAsyncClient.putObject(r -> r.bucket(bucketName).key(key), body);
+    public PutObjectResponse putObjectFromStreamCrt(S3AsyncClient s33CrtAsyncClient, String bucketName, String key) {
 
         // AsyncExampleUtils.randomString() returns a random string up to 100 characters.
         String randomString = AsyncExampleUtils.randomString();
         logger.info("random string to upload: {}: length={}", randomString, randomString.length());
+        InputStream inputStream = new ByteArrayInputStream(randomString.getBytes());
+
+        // Executor required to handle reading from the InputStream on a separate thread so the main upload is not blocked.
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        // Specify `null` for the content length when you don't know the content length.
+        AsyncRequestBody body = AsyncRequestBody.fromInputStream(inputStream, null, executor);
 
-        // Provide the stream of data to be uploaded.
-        body.writeInputStream(new ByteArrayInputStream(randomString.getBytes()));
+        CompletableFuture<PutObjectResponse> responseFuture =
+                s33CrtAsyncClient.putObject(r -> r.bucket(bucketName).key(key), body);
 
         PutObjectResponse response = responseFuture.join(); // Wait for the response.
         logger.info("Object {} uploaded to bucket {}.", key, bucketName);
+        executor.shutdown();
         return response;
     }
 }

javav2/example_code/s3/src/main/java/com/example/s3/async/PutObjectFromStreamAsyncMp.java

@@ -0,0 +1,75 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package com.example.s3.async;
+
+// snippet-start:[s3.java2.async_stream_mp.complete]
+// snippet-start:[s3.java2.async_stream_mp.import]
+
+import com.example.s3.util.AsyncExampleUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+// snippet-end:[s3.java2.async_stream_mp.import]
+
+public class PutObjectFromStreamAsyncMp {
+    private static final Logger logger = LoggerFactory.getLogger(PutObjectFromStreamAsyncMp.class);
+
+    public static void main(String[] args) {
+        String bucketName = "amzn-s3-demo-bucket-" + UUID.randomUUID(); // Change bucket name.
+        String key = UUID.randomUUID().toString();
+
+        AsyncExampleUtils.createBucket(bucketName);
+        try {
+            PutObjectFromStreamAsyncMp example = new PutObjectFromStreamAsyncMp();
+            S3AsyncClient s3AsyncClientMp = S3AsyncClient.builder().multipartEnabled(true).build();
+            PutObjectResponse putObjectResponse = example.putObjectFromStreamMp(s3AsyncClientMp, bucketName, key);
+            logger.info("Object {} etag: {}", key, putObjectResponse.eTag());
+            logger.info("Object {} uploaded to bucket {}.", key, bucketName);
+        } catch (SdkException e) {
+            logger.error(e.getMessage(), e);
+        } finally {
+            AsyncExampleUtils.deleteObject(bucketName, key);
+            AsyncExampleUtils.deleteBucket(bucketName);
+        }
+    }
+
+// snippet-start:[s3.java2.async_stream_mp.main]
+    /**
+     * @param s3AsyncClientMp - To upload content from a stream of unknown size, use can the S3 asynchronous client with multipart enabled.
+     * @param bucketName - The name of the bucket.
+     * @param key - The name of the object.
+     * @return software.amazon.awssdk.services.s3.model.PutObjectResponse - Returns metadata pertaining to the put object operation.
+     */
+    public PutObjectResponse putObjectFromStreamMp(S3AsyncClient s3AsyncClientMp, String bucketName, String key) {
+
+        // AsyncExampleUtils.randomString() returns a random string up to 100 characters.
+        String randomString = AsyncExampleUtils.randomString();
+        logger.info("random string to upload: {}: length={}", randomString, randomString.length());
+        InputStream inputStream = new ByteArrayInputStream(randomString.getBytes());
+
+        // Executor required to handle reading from the InputStream on a separate thread so the main upload is not blocked.
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        // Specify `null` for the content length when you don't know the content length.
+        AsyncRequestBody body = AsyncRequestBody.fromInputStream(inputStream, null, executor);
+
+        CompletableFuture<PutObjectResponse> responseFuture =
+                s3AsyncClientMp.putObject(r -> r.bucket(bucketName).key(key), body);
+
+        PutObjectResponse response = responseFuture.join(); // Wait for the response.
+        logger.info("Object {} uploaded to bucket {}.", key, bucketName);
+        executor.shutdown();
+        return response;
+    }
+}
+// snippet-end:[s3.java2.async_stream_mp.main]
+// snippet-end:[s3.java2.async_stream_mp.complete]

javav2/example_code/s3/src/main/java/com/example/s3/transfermanager/UploadStream.java

@@ -4,18 +4,21 @@
 
 // snippet-start:[s3.tm.java2.upload_stream.complete]
 // snippet-start:[s3.tm.java2.upload_stream.import]
+
 import com.example.s3.util.AsyncExampleUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
-import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.transfer.s3.S3TransferManager;
 import software.amazon.awssdk.transfer.s3.model.CompletedUpload;
 import software.amazon.awssdk.transfer.s3.model.Upload;
 
 import java.io.ByteArrayInputStream;
+import java.io.InputStream;
 import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 // snippet-end:[s3.tm.java2.upload_stream.import]
 
 public class UploadStream {
@@ -41,30 +44,31 @@ public static void main(String[] args) {
 
 // snippet-start:[s3.tm.java2.upload_stream.main]
     /**
-     * @param transferManager - To upload content from a stream of unknown size, use the S3TransferManager based on the AWS CRT-based S3 client.
-     *                       For more information, see https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/transfer-manager.html.
+     * @param transferManager - To upload content from a stream of unknown size, you can use the S3TransferManager based on the AWS CRT-based S3 client.
      * @param bucketName - The name of the bucket.
      * @param key - The name of the object.
      * @return - software.amazon.awssdk.transfer.s3.model.CompletedUpload - The result of the completed upload.
      */
     public CompletedUpload uploadStream(S3TransferManager transferManager, String bucketName, String key) {
 
-        BlockingInputStreamAsyncRequestBody body =
-                AsyncRequestBody.forBlockingInputStream(null); // 'null' indicates a stream will be provided later.
+        // AsyncExampleUtils.randomString() returns a random string up to 100 characters.
+        String randomString = AsyncExampleUtils.randomString();
+        logger.info("random string to upload: {}: length={}", randomString, randomString.length());
+        InputStream inputStream = new ByteArrayInputStream(randomString.getBytes());
+
+        // Executor required to handle reading from the InputStream on a separate thread so the main upload is not blocked.
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        // Specify `null` for the content length when you don't know the content length.
+        AsyncRequestBody body = AsyncRequestBody.fromInputStream(inputStream, null, executor);
 
         Upload upload = transferManager.upload(builder -> builder
                 .requestBody(body)
                 .putObjectRequest(req -> req.bucket(bucketName).key(key))
                 .build());
 
-        // AsyncExampleUtils.randomString() returns a random string up to 100 characters.
-        String randomString = AsyncExampleUtils.randomString();
-        logger.info("random string to upload: {}: length={}", randomString, randomString.length());
-
-        // Provide the stream of data to be uploaded.
-        body.writeInputStream(new ByteArrayInputStream(randomString.getBytes()));
-
-        return upload.completionFuture().join();
+        CompletedUpload completedUpload = upload.completionFuture().join();
+        executor.shutdown();
+        return completedUpload;
     }
 }
 // snippet-end:[s3.tm.java2.upload_stream.main]

javav2/example_code/s3/src/test/java/com/example/s3/async/AsyncTests.java

@@ -10,6 +10,7 @@
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 
 import java.util.UUID;
@@ -34,9 +35,17 @@ void tearDown() {
 
     @Test
     @Tag("IntegrationTest")
-    void putObjectFromStream() {
+    void putObjectFromStreamTest() {
         PutObjectFromStreamAsync example = new PutObjectFromStreamAsync();
-        PutObjectResponse putObjectResponse = example.putObjectFromStream(AsyncExampleUtils.client, bucketName, key);
+        PutObjectResponse putObjectResponse = example.putObjectFromStreamCrt(AsyncExampleUtils.client, bucketName, key);
+        Assertions.assertNotNull(putObjectResponse.eTag());
+    }
+
+    @Test
+    @Tag("IntegrationTest")
+    void putObjectFromStreamMpTest() {
+        PutObjectFromStreamAsyncMp example = new PutObjectFromStreamAsyncMp();
+        PutObjectResponse putObjectResponse = example.putObjectFromStreamMp(S3AsyncClient.builder().multipartEnabled(true).build(), bucketName, key);
         Assertions.assertNotNull(putObjectResponse.eTag());
     }
 }