Plain transplant of existing AsyncResponseTransformer

Author: rtyley

src/main/java/com/madgag/aws/sdk/async/responsebytes/Main.java

@@ -2,9 +2,9 @@
 
 import com.google.common.hash.HashCode;
 import com.google.common.hash.Hashing;
+import com.madgag.aws.sdk.async.responsebytes.awssdk.core.async.AsyncResponseTransformerAlternative;
 import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
 import software.amazon.awssdk.core.ResponseBytes;
-import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.http.crt.AwsCrtAsyncHttpClient;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
@@ -23,7 +23,7 @@ public static void main(String[] args) throws ExecutionException, InterruptedExc
             KnownS3Object knownS3Object = KnownS3Object.BIG;
             ResponseBytes<GetObjectResponse> responseBytes = s3Client.getObject(
                     GetObjectRequest.builder().bucket(knownS3Object.bucket()).key(knownS3Object.key()).build(),
-                    AsyncResponseTransformer.toBytes()
+                    AsyncResponseTransformerAlternative.toBytes()
             ).get();
             System.out.println(responseBytes.response().contentLength());
             HashCode hashOfDownloadedData = Hashing.sha256().hashBytes(responseBytes.asByteBuffer());

src/main/java/com/madgag/aws/sdk/async/responsebytes/awssdk/core/async/AsyncResponseTransformerAlternative.java

@@ -0,0 +1,11 @@
+package com.madgag.aws.sdk.async.responsebytes.awssdk.core.async;
+
+import com.madgag.aws.sdk.async.responsebytes.awssdk.core.internal.async.ByteArrayAsyncResponseTransformerAlternative;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+
+public class AsyncResponseTransformerAlternative {
+    public static <ResponseT> AsyncResponseTransformer<ResponseT, ResponseBytes<ResponseT>> toBytes() {
+        return new ByteArrayAsyncResponseTransformerAlternative<>();
+    }
+}

src/main/java/com/madgag/aws/sdk/async/responsebytes/awssdk/core/internal/async/ByteArrayAsyncResponseTransformerAlternative.java

@@ -0,0 +1,106 @@
+/*
+ * 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 com.madgag.aws.sdk.async.responsebytes.awssdk.core.internal.async;
+
+import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.core.async.SdkPublisher;
+import software.amazon.awssdk.utils.BinaryUtils;
+
+/**
+ * Implementation of {@link AsyncResponseTransformer} that dumps content into a byte array and supports further
+ * conversions into types, like strings.
+ *
+ * This can be created with static methods on {@link AsyncResponseTransformer}.
+ *
+ * @param <ResponseT> Pojo response type.
+ * @see AsyncResponseTransformer#toBytes()
+ */
+@SdkInternalApi
+public final class ByteArrayAsyncResponseTransformerAlternative<ResponseT> implements
+        AsyncResponseTransformer<ResponseT, ResponseBytes<ResponseT>> {
+
+    private volatile CompletableFuture<byte[]> cf;
+    private volatile ResponseT response;
+
+    @Override
+    public CompletableFuture<ResponseBytes<ResponseT>> prepare() {
+        cf = new CompletableFuture<>();
+        return cf.thenApply(arr -> ResponseBytes.fromByteArray(response, arr));
+    }
+
+    @Override
+    public void onResponse(ResponseT response) {
+        this.response = response;
+    }
+
+    @Override
+    public void onStream(SdkPublisher<ByteBuffer> publisher) {
+        publisher.subscribe(new BaosSubscriber(cf));
+    }
+
+    @Override
+    public void exceptionOccurred(Throwable throwable) {
+        cf.completeExceptionally(throwable);
+    }
+
+    static class BaosSubscriber implements Subscriber<ByteBuffer> {
+        private final CompletableFuture<byte[]> resultFuture;
+
+        private ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+        private Subscription subscription;
+
+        BaosSubscriber(CompletableFuture<byte[]> resultFuture) {
+            this.resultFuture = resultFuture;
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (this.subscription != null) {
+                s.cancel();
+                return;
+            }
+            this.subscription = s;
+            subscription.request(Long.MAX_VALUE);
+        }
+
+        @Override
+        public void onNext(ByteBuffer byteBuffer) {
+            invokeSafely(() -> baos.write(BinaryUtils.copyBytesFrom(byteBuffer)));
+            subscription.request(1);
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+            baos = null;
+            resultFuture.completeExceptionally(throwable);
+        }
+
+        @Override
+        public void onComplete() {
+            resultFuture.complete(baos.toByteArray());
+        }
+    }
+}