Added AsyncRequestBody.forBlockingOutputStream (#3565)

Added AsyncRequestBody.forBlockingOutputStream, allowing streaming operation requests to be written to like an output stream.

Example usage:
```java
BlockingOutputStreamAsyncRequestBody body = AsyncRequestBody.forBlockingOutputStream(5L);
CompletableFuture<?> responseFuture =
    client.streamingInputOperation(StreamingInputOperationRequest.builder().build(), body);

try (OutputStream stream = body.outputStream()) { // TODO: Test writing too much data and too little data
    stream.write("Hello".getBytes(StandardCharsets.UTF_8));
}
responseFuture.join();
```

Other changes made to support this change:
1. Added `OutputStreamPublisher` to utils. This is a `Publisher<ByteBuffer>` that can be written to like it's an output stream. Writes will block when there is no available downstream demand.
1. Added `CancellableOutputStream` to utils. This is an output stream that can be cancelled. This is useful for cancelling the upload via the output stream.

The following changes appear in this PR as well as #3563:
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-2c6b2b3.json

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

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

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.core.async;
 
 import java.io.File;
+import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
@@ -160,6 +161,40 @@ static AsyncRequestBody fromByteBuffer(ByteBuffer byteBuffer) {
         return fromBytes(BinaryUtils.copyAllBytesFrom(byteBuffer));
     }
 
+    /**
+     * 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.
+     *
+     * <p>The caller is responsible for calling {@link OutputStream#close()} on the {@link #outputStream()} when writing is
+     * complete.
+     *
+     * <p><b>Example Usage</b>
+     * <p>
+     * {@snippet :
+     *     S3AsyncClient s3 = S3AsyncClient.create(); // Use one client for your whole application!
+     *
+     *     byte[] dataToSend = "Hello".getBytes(StandardCharsets.UTF_8);
+     *     long lengthOfDataToSend = dataToSend.length();
+     *
+     *     // Start the operation
+     *     BlockingInputStreamAsyncRequestBody body =
+     *         AsyncRequestBody.forBlockingOutputStream(lengthOfDataToSend);
+     *     CompletableFuture<PutObjectResponse> responseFuture =
+     *         s3.putObject(r -> r.bucket("bucketName").key("key"), body);
+     *
+     *     // Write the input stream to the running operation
+     *     try (CancellableOutputStream outputStream = body.outputStream()) {
+     *         outputStream.write(dataToSend);
+     *     }
+     *
+     *     // Wait for the service to respond.
+     *     PutObjectResponse response = responseFuture.join();
+     * }
+     */
+    static BlockingOutputStreamAsyncRequestBody forBlockingOutputStream(Long contentLength) {
+        return new BlockingOutputStreamAsyncRequestBody(contentLength);
+    }
+
     /**
      * Creates a {@link AsyncRequestBody} with no content.
      *

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

@@ -0,0 +1,102 @@
+/*
+ * 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.OutputStream;
+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.util.NoopSubscription;
+import software.amazon.awssdk.utils.CancellableOutputStream;
+import software.amazon.awssdk.utils.async.OutputStreamPublisher;
+
+/**
+ * An implementation of {@link AsyncRequestBody} that allows performing a blocking write of an output stream to a downstream
+ * service.
+ *
+ * <p>The caller is responsible for calling {@link OutputStream#close()} on the {@link #outputStream()} when writing is
+ * complete.
+ *
+ * @see AsyncRequestBody#forBlockingOutputStream(Long)
+ */
+@SdkPublicApi
+public final class BlockingOutputStreamAsyncRequestBody implements AsyncRequestBody {
+    private final OutputStreamPublisher delegate = new OutputStreamPublisher();
+    private final CountDownLatch subscribedLatch = new CountDownLatch(1);
+    private final AtomicBoolean subscribeCalled = new AtomicBoolean(false);
+    private final Long contentLength;
+    private final Duration subscribeTimeout;
+
+    BlockingOutputStreamAsyncRequestBody(Long contentLength) {
+        this(contentLength, Duration.ofSeconds(10));
+    }
+
+    BlockingOutputStreamAsyncRequestBody(Long contentLength, Duration subscribeTimeout) {
+        this.contentLength = contentLength;
+        this.subscribeTimeout = subscribeTimeout;
+    }
+
+    /**
+     * Return an output stream to which blocking writes can be made to the downstream service.
+     *
+     * <p>This method will block the calling thread until the SDK is connected to the service. This means that this request body
+     * should usually be passed to the SDK before this method is called.
+     *
+     * <p>You can invoke {@link CancellableOutputStream#cancel()} to cancel any blocked write calls to the downstream service
+     * (and mark the stream as failed).
+     */
+    public CancellableOutputStream outputStream() {
+        waitForSubscriptionIfNeeded();
+        return delegate;
+    }
+
+    @Override
+    public Optional<Long> contentLength() {
+        return Optional.ofNullable(contentLength);
+    }
+
+    @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.forBlockingOutputStream does not "
+                                                   + "support retries."));
+        }
+    }
+
+    private void waitForSubscriptionIfNeeded() {
+        try {
+            long timeoutSeconds = subscribeTimeout.getSeconds();
+            if (!subscribedLatch.await(timeoutSeconds, TimeUnit.SECONDS)) {
+                throw new IllegalStateException("The service request was not made within " + timeoutSeconds + " seconds of "
+                                                + "outputStream being invoked. Make sure to invoke the service request "
+                                                + "BEFORE invoking outputStream if your caller is single-threaded.");
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("Interrupted while waiting for subscription.", e);
+        }
+    }
+}

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

@@ -0,0 +1,80 @@
+/*
+ * 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 static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static software.amazon.awssdk.utils.async.ByteBufferStoringSubscriber.TransferResult.END_OF_STREAM;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import software.amazon.awssdk.utils.CancellableOutputStream;
+import software.amazon.awssdk.utils.async.ByteBufferStoringSubscriber;
+import software.amazon.awssdk.utils.async.StoringSubscriber;
+
+class BlockingOutputStreamAsyncRequestBodyTest {
+    @Test
+    public void outputStream_waitsForSubscription() throws IOException {
+        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
+        try {
+            BlockingOutputStreamAsyncRequestBody requestBody =
+                AsyncRequestBody.forBlockingOutputStream(0L);
+            executor.schedule(() -> requestBody.subscribe(new StoringSubscriber<>(1)), 100, MILLISECONDS);
+            try (OutputStream stream = requestBody.outputStream()) {
+                stream.write('h');
+            }
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+
+    @Test
+    @Timeout(10)
+    public void outputStream_failsIfSubscriptionNeverComes()  {
+        BlockingOutputStreamAsyncRequestBody requestBody =
+            new BlockingOutputStreamAsyncRequestBody(0L, Duration.ofSeconds(1));
+        assertThatThrownBy(requestBody::outputStream).hasMessageContaining("The service request was not made");
+    }
+
+    @Test
+    public void outputStream_writesToSubscriber() throws IOException {
+        BlockingOutputStreamAsyncRequestBody requestBody =
+            AsyncRequestBody.forBlockingOutputStream(0L);
+        ByteBufferStoringSubscriber subscriber = new ByteBufferStoringSubscriber(4);
+        requestBody.subscribe(subscriber);
+
+        CancellableOutputStream outputStream = requestBody.outputStream();
+        outputStream.write(0);
+        outputStream.write(1);
+        outputStream.close();
+
+        ByteBuffer out = ByteBuffer.allocate(4);
+        assertThat(subscriber.transferTo(out)).isEqualTo(END_OF_STREAM);
+        out.flip();
+
+        assertThat(out.remaining()).isEqualTo(2);
+        assertThat(out.get()).isEqualTo((byte) 0);
+        assertThat(out.get()).isEqualTo((byte) 1);
+    }
+
+}

services/s3/src/it/java/software/amazon/awssdk/services/s3/AsyncInOutStreamIntegrationTest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.services.s3;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.CompletableFuture;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.async.BlockingOutputStreamAsyncRequestBody;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.testutils.service.S3BucketUtils;
+
+/**
+ * Test InputStream and OutputStream operations on AsyncRequestBody and AsyncResponseTransformer against S3.
+ */
+public class AsyncInOutStreamIntegrationTest extends S3IntegrationTestBase {
+
+    private static final String BUCKET = S3BucketUtils.temporaryBucketName("async-in-out-stream");
+
+    /**
+     * Creates all the test resources for the tests.
+     */
+    @BeforeClass
+    public static void createResources() {
+        createBucket(BUCKET);
+    }
+
+    /**
+     * Releases all resources created in this test.
+     */
+    @AfterClass
+    public static void tearDown() {
+        deleteBucketAndAllContents(BUCKET);
+    }
+
+    @Test
+    public void largeFilePutGet() throws IOException {
+        long length = 4 * 1024 * 1024; // 4 MB
+        BlockingOutputStreamAsyncRequestBody body = AsyncRequestBody.forBlockingOutputStream(length);
+        CompletableFuture<?> response = s3Async.putObject(r -> r.bucket(BUCKET).key("foo"), body);
+
+        try (OutputStream os = body.outputStream()) {
+            for (int i = 0; i < length; i++) {
+                os.write(i % 255);
+            }
+        }
+
+        response.join();
+
+        try (ResponseInputStream<GetObjectResponse> is =
+                 s3Async.getObject(r -> r.bucket(BUCKET).key("foo"),
+                                   AsyncResponseTransformer.toBlockingInputStream())
+                        .join()) {
+
+            assertThat(is.response().contentLength()).isEqualTo(length);
+
+            for (int i = 0; i < length; i++) {
+                assertThat(is.read()).isEqualTo(i % 255);
+            }
+        }
+    }
+}