Added AsyncResponseBody.fromInputStream and AsyncRequestBody.forBlockingInputStream. (#3563)

Added AsyncResponseBody.fromInputStream and AsyncRequestBody.forBlockingInputStream, allowing streaming operation requests to be written to like an input stream.

Example usage with retry support, without blocking calling thread:

```java
String data = "Hello, world!";
ExecutorService executor = Executors.newCachedThreadPool();
InputStream data = new StringInputStream("Hello, world!");

s3.putObject(r -> r.bucket("foo").key("bar"),
             AsyncRequestBody.fromInputStream(data, data.length(), executor))
  .join();
```

Example usage without retry support, with blocking calling thread:
```java
String data = "Hello, world!";
BlockingInputStreamAsyncRequestBody body = AsyncRequestBody.forBlockingInputStream(data.length());
CompletableFuture<?> responseFuture = s3.putObject(r -> r.bucket("foo").key("bar"), body);
body.writeInputStream(new StringInputStream(data));
responseFuture.join();
```

Other changes made to support this change:
1. Added `InputStreamConsumingPublisher`, a `Publisher<ByteBuffer>` that can read from an `InputStream`.

These supporting changes were also made, and are also being reviewed in #3565:
1. Updated `SimplePublisher` to return more useful and predictable error messages when invoked after `onComplete`, `onError` or `cancel` has been called.
1. Added `CompletableFutureUtils.joinInterruptibly`, which behaves equivalent to executing `join()`, but allows the calling thread to still respond to interrupts.
1. Added `CompletableFutureUtils.joinInterruptiblyIgnoringFailures`, which behaves equivalent to executing `join()`, but allows the calling thread to still respond to interrupts, and ignores exceptions that are thrown.

Author: millems

.changes/next-release/feature-AWSSDKforJavav2-303765f.json

@@ -0,0 +1,6 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "contributor": "", 
+    "type": "feature", 
+    "description": "Added AsyncResponseBody.fromInputStream and AsyncRequestBody.forBlockingInputStream, allowing streaming operation requests to be written to like an input stream."
+}

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

@@ -16,17 +16,20 @@
 package software.amazon.awssdk.core.async;
 
 import java.io.File;
+import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.core.internal.async.ByteArrayAsyncRequestBody;
 import software.amazon.awssdk.core.internal.async.FileAsyncRequestBody;
+import software.amazon.awssdk.core.internal.async.InputStreamWithExecutorAsyncRequestBody;
 import software.amazon.awssdk.core.internal.util.Mimetype;
 import software.amazon.awssdk.utils.BinaryUtils;
 
@@ -161,6 +164,44 @@ static AsyncRequestBody fromByteBuffer(ByteBuffer byteBuffer) {
         return fromBytes(BinaryUtils.copyAllBytesFrom(byteBuffer));
     }
 
+    /**
+     * Creates a {@link AsyncRequestBody} from a {@link InputStream}.
+     *
+     * <p>An {@link ExecutorService} is required in order to perform the blocking data reads, to prevent blocking the
+     * non-blocking event loop threads owned by the SDK.
+     */
+    static AsyncRequestBody fromInputStream(InputStream inputStream, Long contentLength, ExecutorService executor) {
+        return new InputStreamWithExecutorAsyncRequestBody(inputStream, contentLength, executor);
+    }
+
+    /**
+     * Creates a {@link BlockingInputStreamAsyncRequestBody} to use for writing an input stream to the downstream service.
+     *
+     * <p><b>Example Usage</b>
+     *
+     * <pre>
+     *     S3Client s3 = ...;
+     *
+     *     InputStream streamToWrite = ...;
+     *     long streamToWriteLength = ...;
+     *
+     *     // Start the operation
+     *     BlockingInputStreamAsyncRequestBody body =
+     *         AsyncRequestBody.forBlockingInputStream(streamToWriteLength);
+     *     CompletableFuture<PutObjectResponse> responseFuture =
+     *         s3.putObject(r -> r.bucket("bucketName").key("key"), body);
+     *
+     *     // Write the input stream to the running operation
+     *     body.writeInputStream(streamToWrite);
+     *
+     *     // Wait for the service to respond.
+     *     PutObjectResponse response = responseFuture.join();
+     * </pre>
+     */
+    static BlockingInputStreamAsyncRequestBody forBlockingInputStream(Long contentLength) {
+        return new BlockingInputStreamAsyncRequestBody(contentLength);
+    }
+
     /**
      * Creates a {@link BlockingOutputStreamAsyncRequestBody} to use for writing to the downstream service as if it's an output
      * stream. Retries are not supported for this request body.

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

@@ -0,0 +1,115 @@
+/*
+ * 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.core.async;
+
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.reactivestreams.Subscriber;
+import software.amazon.awssdk.annotations.SdkPublicApi;
+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.async.InputStreamConsumingPublisher;
+
+/**
+ * An implementation of {@link AsyncRequestBody} that allows performing a blocking write of an input stream to a downstream
+ * service.
+ *
+ * <p>See {@link AsyncRequestBody#forBlockingInputStream(Long)}.
+ */
+@SdkPublicApi
+public final class BlockingInputStreamAsyncRequestBody implements AsyncRequestBody {
+    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(Long contentLength, Duration subscribeTimeout) {
+        this.contentLength = contentLength;
+        this.subscribeTimeout = subscribeTimeout;
+    }
+
+    @Override
+    public Optional<Long> contentLength() {
+        return Optional.ofNullable(contentLength);
+    }
+
+    /**
+     * Block the calling thread and write the provided input stream to the downstream service.
+     *
+     * <p>This method will block the calling thread immediately. This means that this request body should usually be passed to
+     * the SDK before this method is called.
+     *
+     * <p>This method will return the amount of data written when the entire input stream has been written. This will throw an
+     * exception if writing the input stream has failed.
+     *
+     * <p>You can invoke {@link #cancel()} to cancel any blocked write calls to the downstream service (and mark the stream as
+     * failed).
+     */
+    public long writeInputStream(InputStream inputStream) {
+        try {
+            waitForSubscriptionIfNeeded();
+            if (contentLength != null) {
+                return delegate.doBlockingWrite(new SdkLengthAwareInputStream(inputStream, contentLength));
+            }
+
+            return delegate.doBlockingWrite(inputStream);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            delegate.cancel();
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * Cancel any running write (and mark the stream as failed).
+     */
+    public void cancel() {
+        delegate.cancel();
+    }
+
+    @Override
+    public void subscribe(Subscriber<? super ByteBuffer> s) {
+        if (subscribeCalled.compareAndSet(false, true)) {
+            delegate.subscribe(s);
+            subscribedLatch.countDown();
+        } else {
+            s.onSubscribe(new NoopSubscription(s));
+            s.onError(NonRetryableException.create("A retry was attempted, but AsyncRequestBody.forBlockingInputStream does not "
+                                                   + "support retries. Consider using AsyncRequestBody.fromInputStream with an "
+                                                   + "input stream that supports mark/reset to get retry support."));
+        }
+    }
+
+    private void waitForSubscriptionIfNeeded() throws InterruptedException {
+        long timeoutSeconds = subscribeTimeout.getSeconds();
+        if (!subscribedLatch.await(timeoutSeconds, TimeUnit.SECONDS)) {
+            throw new IllegalStateException("The service request was not made within " + timeoutSeconds + " seconds of "
+                                            + "doBlockingWrite being invoked. Make sure to invoke the service request "
+                                            + "BEFORE invoking doBlockingWrite if your caller is single-threaded.");
+        }
+    }
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/exception/NonRetryableException.java

@@ -45,6 +45,14 @@ public static Builder builder() {
         return new BuilderImpl();
     }
 
+    public static NonRetryableException create(String message) {
+        return builder().message(message).build();
+    }
+
+    public static NonRetryableException create(String message, Throwable cause) {
+        return builder().message(message).cause(cause).build();
+    }
+
     public interface Builder extends SdkClientException.Builder {
         @Override
         Builder message(String message);

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

@@ -0,0 +1,127 @@
+/*
+ * 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.core.internal.async;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Optional;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.reactivestreams.Subscriber;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.annotations.SdkTestInternalApi;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody;
+import software.amazon.awssdk.core.exception.NonRetryableException;
+import software.amazon.awssdk.core.internal.util.NoopSubscription;
+import software.amazon.awssdk.utils.IoUtils;
+import software.amazon.awssdk.utils.Logger;
+
+/**
+ * A {@link AsyncRequestBody} that allows reading data off of an {@link InputStream} using a background
+ * {@link ExecutorService}.
+ * <p>
+ * Created via {@link AsyncRequestBody#fromInputStream(InputStream, Long, ExecutorService)}.
+ */
+@SdkInternalApi
+public class InputStreamWithExecutorAsyncRequestBody implements AsyncRequestBody {
+    private static final Logger log = Logger.loggerFor(InputStreamWithExecutorAsyncRequestBody.class);
+
+    private final Object subscribeLock = new Object();
+    private final InputStream inputStream;
+    private final Long contentLength;
+    private final ExecutorService executor;
+
+    private Future<?> writeFuture;
+
+    public InputStreamWithExecutorAsyncRequestBody(InputStream inputStream,
+                                                   Long contentLength,
+                                                   ExecutorService executor) {
+        this.inputStream = inputStream;
+        this.contentLength = contentLength;
+        this.executor = executor;
+        IoUtils.markStreamWithMaxReadLimit(inputStream);
+    }
+
+    @Override
+    public Optional<Long> contentLength() {
+        return Optional.ofNullable(contentLength);
+    }
+
+    @Override
+    public void subscribe(Subscriber<? super ByteBuffer> s) {
+        // Each subscribe cancels the previous subscribe.
+        synchronized (subscribeLock) {
+            try {
+                if (writeFuture != null) {
+                    writeFuture.cancel(true);
+                    waitForCancellation(writeFuture); // Wait for the cancellation
+                    tryReset(inputStream);
+                }
+
+                BlockingInputStreamAsyncRequestBody delegate = AsyncRequestBody.forBlockingInputStream(contentLength);
+                writeFuture = executor.submit(() -> doBlockingWrite(delegate));
+                delegate.subscribe(s);
+            } catch (Throwable t) {
+                s.onSubscribe(new NoopSubscription(s));
+                s.onError(t);
+            }
+        }
+    }
+
+    private void tryReset(InputStream inputStream) {
+        try {
+            inputStream.reset();
+        } catch (IOException e) {
+            String message = "Request cannot be retried, because the request stream could not be reset.";
+            throw NonRetryableException.create(message, e);
+        }
+    }
+
+    @SdkTestInternalApi
+    public Future<?> activeWriteFuture() {
+        synchronized (subscribeLock) {
+            return writeFuture;
+        }
+    }
+
+    private void doBlockingWrite(BlockingInputStreamAsyncRequestBody asyncRequestBody) {
+        try {
+            asyncRequestBody.writeInputStream(inputStream);
+        } catch (Throwable t) {
+            log.debug(() -> "Encountered error while writing input stream to service.", t);
+            throw t;
+        }
+    }
+
+    private void waitForCancellation(Future<?> writeFuture) {
+        try {
+            writeFuture.get(10, TimeUnit.SECONDS);
+        } catch (ExecutionException | CancellationException e) {
+            // Expected - we cancelled.
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        } catch (TimeoutException e) {
+            throw new IllegalStateException("Timed out waiting to reset the input stream.", e);
+        }
+    }
+}