Allows users to configure subscribeTimeout for BlockingInputStreamAsyncRequestBody (aws#5020)

* allow user to configure subscribe timeout for input stream

* Fix compilation error

* Address feedback

---------

Co-authored-by: Ben Arnao <benarnao@amazon.com>

Author: 2 people

.changes/next-release/feature-AWSS3-1be6ddd.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS S3",
+    "contributor": "benarnao",
+    "description": "allow user to configure subscriber timeout for input stream"
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/AsyncRequestBody.java

@@ -384,6 +384,8 @@ static AsyncRequestBody fromInputStream(Consumer<AsyncRequestBodyFromInputStream
     /**
      * Creates a {@link BlockingInputStreamAsyncRequestBody} to use for writing an input stream to the downstream service.
      *
+     * <p>By default, it will time out if streaming hasn't started within 10 seconds, and you can configure the timeout
+     * via {@link BlockingInputStreamAsyncRequestBody#builder()}
      * <p><b>Example Usage</b>
      *
      * <p>
@@ -408,7 +410,9 @@ static AsyncRequestBody fromInputStream(Consumer<AsyncRequestBodyFromInputStream
      * }
      */
     static BlockingInputStreamAsyncRequestBody forBlockingInputStream(Long contentLength) {
-        return new BlockingInputStreamAsyncRequestBody(contentLength);
+        return BlockingInputStreamAsyncRequestBody.builder()
+                                                  .contentLength(contentLength)
+                                                  .build();
     }
 
     /**

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/BlockingInputStreamAsyncRequestBody.java

@@ -27,6 +27,7 @@
 import software.amazon.awssdk.core.exception.NonRetryableException;
 import software.amazon.awssdk.core.internal.io.SdkLengthAwareInputStream;
 import software.amazon.awssdk.core.internal.util.NoopSubscription;
+import software.amazon.awssdk.utils.Validate;
 import software.amazon.awssdk.utils.async.InputStreamConsumingPublisher;
 
 /**
@@ -37,19 +38,25 @@
  */
 @SdkPublicApi
 public final class BlockingInputStreamAsyncRequestBody implements AsyncRequestBody {
+    private static final Duration DEFAULT_SUBSCRIBE_TIMEOUT = Duration.ofSeconds(10);
     private final InputStreamConsumingPublisher delegate = new InputStreamConsumingPublisher();
     private final CountDownLatch subscribedLatch = new CountDownLatch(1);
     private final AtomicBoolean subscribeCalled = new AtomicBoolean(false);
     private final Long contentLength;
     private final Duration subscribeTimeout;
 
-    BlockingInputStreamAsyncRequestBody(Long contentLength) {
-        this(contentLength, Duration.ofSeconds(10));
+    BlockingInputStreamAsyncRequestBody(Builder builder) {
+        this.contentLength = builder.contentLength;
+        this.subscribeTimeout = Validate.isPositiveOrNull(builder.subscribeTimeout, "subscribeTimeout") != null ?
+                                builder.subscribeTimeout :
+                                DEFAULT_SUBSCRIBE_TIMEOUT;
     }
 
-    BlockingInputStreamAsyncRequestBody(Long contentLength, Duration subscribeTimeout) {
-        this.contentLength = contentLength;
-        this.subscribeTimeout = subscribeTimeout;
+    /**
+     * Creates a default builder for {@link BlockingInputStreamAsyncRequestBody}.
+     */
+    public static Builder builder() {
+        return new Builder();
     }
 
     @Override
@@ -112,4 +119,41 @@ private void waitForSubscriptionIfNeeded() throws InterruptedException {
                                             + "BEFORE invoking doBlockingWrite if your caller is single-threaded.");
         }
     }
+
+    public static final class Builder {
+        private Duration subscribeTimeout;
+        private Long contentLength;
+
+        private Builder() {
+        }
+
+        /**
+         * Defines how long it should wait for this AsyncRequestBody to be subscribed (to start streaming) before timing out.
+         * By default, it's 10 seconds.
+         *
+         * <p>You may want to increase it if the request may not be executed right away.
+         *
+         * @param subscribeTimeout the timeout
+         * @return Returns a reference to this object so that method calls can be chained together.
+         */
+        public Builder subscribeTimeout(Duration subscribeTimeout) {
+            this.subscribeTimeout = subscribeTimeout;
+            return this;
+        }
+
+        /**
+         * The content length of the output stream.
+         *
+         * @param contentLength the content length
+         * @return Returns a reference to this object so that method calls can be chained together.
+         */
+        public Builder contentLength(Long contentLength) {
+            this.contentLength = contentLength;
+            return this;
+        }
+
+        public BlockingInputStreamAsyncRequestBody build() {
+            return new BlockingInputStreamAsyncRequestBody(this);
+        }
+    }
 }

core/sdk-core/src/test/java/software/amazon/awssdk/core/async/BlockingInputStreamAsyncRequestBodyTest.java

@@ -47,9 +47,9 @@ public void doBlockingWrite_waitsForSubscription() {
 
     @Test
     @Timeout(10)
-    public void doBlockingWrite_failsIfSubscriptionNeverComes()  {
+    public void doBlockingWrite_overrideSubscribeTimeout_failsIfSubscriptionNeverComes()  {
         BlockingInputStreamAsyncRequestBody requestBody =
-            new BlockingInputStreamAsyncRequestBody(0L, Duration.ofSeconds(1));
+            BlockingInputStreamAsyncRequestBody.builder().contentLength(0L).subscribeTimeout(Duration.ofSeconds(1)).build();
         assertThatThrownBy(() -> requestBody.writeInputStream(new StringInputStream("")))
             .hasMessageContaining("The service request was not made");
     }