Implement getObject in S3CrtAsyncClient

Author: zoewangg

pom.xml

@@ -112,7 +112,7 @@
         <rxjava.version>2.1.9</rxjava.version>
         <commons-codec.verion>1.10</commons-codec.verion>
         <jmh.version>1.21</jmh.version>
-        <awscrt.version>0.10.9</awscrt.version>
+        <awscrt.version>1.0.0-SNAPSHOT</awscrt.version>
 
         <!-- Test dependencies -->
         <junit.version>4.12</junit.version>
@@ -163,7 +163,7 @@
         <unitils.version>3.3</unitils.version>
 
         <!-- Reactive Streams version -->
-        <reactive-streams.version>1.0.2</reactive-streams.version>
+        <reactive-streams.version>1.0.3</reactive-streams.version>
 
         <skip.unit.tests>${skipTests}</skip.unit.tests>
     </properties>

services/s3/pom.xml

@@ -111,5 +111,10 @@
             <artifactId>wiremock</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams-tck</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/CrtResponseDataConsumerAdapter.java

@@ -0,0 +1,85 @@
+/*
+ * 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.internal;
+
+import com.amazonaws.s3.ResponseDataConsumer;
+import com.amazonaws.s3.model.GetObjectOutput;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.SdkStandardLogger;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.crt.CrtRuntimeException;
+import software.amazon.awssdk.crt.http.HttpHeader;
+import software.amazon.awssdk.http.HttpStatusFamily;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.utils.Logger;
+
+/**
+ * Adapt the SDK API {@link AsyncResponseTransformer} to the CRT API {@link ResponseDataConsumer}.
+ */
+@SdkInternalApi
+public final class CrtResponseDataConsumerAdapter<ReturnT> implements ResponseDataConsumer<GetObjectOutput> {
+    private static final Logger log = Logger.loggerFor(CrtResponseDataConsumerAdapter.class);
+    private final AsyncResponseTransformer<GetObjectResponse, ReturnT> transformer;
+    private final CompletableFuture<ReturnT> future;
+    private final S3CrtDataPublisher publisher;
+
+    public CrtResponseDataConsumerAdapter(AsyncResponseTransformer<GetObjectResponse, ReturnT> transformer) {
+        this.transformer = transformer;
+        this.future = transformer.prepare();
+        this.publisher = new S3CrtDataPublisher();
+    }
+
+    public CompletableFuture<ReturnT> transformerFuture() {
+        return future;
+    }
+
+    @Override
+    public void onResponseHeaders(int statusCode, HttpHeader[] headers) {
+        if (HttpStatusFamily.of(statusCode) == HttpStatusFamily.SUCCESSFUL) {
+            SdkStandardLogger.REQUEST_LOGGER.debug(() -> "Received successful response: " + statusCode);
+        } else {
+            SdkStandardLogger.REQUEST_LOGGER.debug(() -> "Received error response: " + statusCode);
+        }
+    }
+
+    @Override
+    public void onResponse(GetObjectOutput output) {
+        GetObjectResponse response = S3CrtUtils.adaptGetObjectOutput(output);
+        transformer.onResponse(response);
+        transformer.onStream(publisher);
+    }
+
+    @Override
+    public void onResponseData(ByteBuffer byteBuffer) {
+        log.debug(() -> "Received data of size " + byteBuffer.remaining());
+        publisher.deliverData(byteBuffer);
+    }
+
+    @Override
+    public void onException(CrtRuntimeException e) {
+        log.debug(() -> "An error occurred ", e);
+        transformer.exceptionOccurred(e);
+        publisher.notifyError(e);
+    }
+
+    @Override
+    public void onFinished() {
+        log.debug(() -> "Finished streaming ");
+        publisher.notifyStreamingFinished();
+    }
+}

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/DefaultS3CrtAsyncClient.java

@@ -19,24 +19,26 @@
 import static software.amazon.awssdk.services.s3.internal.S3CrtUtils.createCrtCredentialsProvider;
 
 import com.amazonaws.s3.S3NativeClient;
+import java.util.concurrent.CompletableFuture;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.crt.auth.credentials.CredentialsProvider;
 import software.amazon.awssdk.services.s3.S3CrtAsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 
 @SdkInternalApi
 public final class DefaultS3CrtAsyncClient implements S3CrtAsyncClient {
     private final S3NativeClient s3NativeClient;
     private final S3NativeClientConfiguration configuration;
 
-    private final CredentialsProvider credentialsProvider;
-
-
     public DefaultS3CrtAsyncClient(DefaultS3CrtClientBuilder builder) {
-        this.credentialsProvider = createCrtCredentialsProvider(builder.credentialsProvider());
+        CredentialsProvider credentialsProvider =  builder.credentialsProvider() == null ? null :
+                                                   createCrtCredentialsProvider(builder.credentialsProvider());
 
         this.configuration = S3NativeClientConfiguration.builder()
                                                         .credentialsProvider(credentialsProvider)
-                                                        .signingRegion(builder.region().id())
+                                                        .signingRegion(builder.region() == null ? null : builder.region().id())
                                                         .partSizeBytes(builder.partSizeBytes())
                                                         .maxThroughputGbps(builder.maxThroughputGbps())
                                                         .build();
@@ -48,7 +50,29 @@ public DefaultS3CrtAsyncClient(DefaultS3CrtClientBuilder builder) {
                                                  configuration.maxThroughputGbps());
     }
 
+    @Override
+    public <ReturnT> CompletableFuture<ReturnT> getObject(
+        GetObjectRequest getObjectRequest, AsyncResponseTransformer<GetObjectResponse, ReturnT> asyncResponseTransformer) {
+
+        CompletableFuture<ReturnT> future = new CompletableFuture<>();
+        com.amazonaws.s3.model.GetObjectRequest crtGetObjectRequest = S3CrtUtils.adaptGetObjectRequest(getObjectRequest);
+        CrtResponseDataConsumerAdapter<ReturnT> adapter = new CrtResponseDataConsumerAdapter<>(asyncResponseTransformer);
 
+        CompletableFuture<ReturnT> adapterFuture = adapter.transformerFuture();
+
+        s3NativeClient.getObject(crtGetObjectRequest, adapter);
+
+        adapterFuture.whenComplete((r, t) -> {
+            if (t == null) {
+                future.complete(r);
+            } else {
+                future.completeExceptionally(t);
+            }
+            // TODO: Offload to future completion thread
+        });
+
+        return future;
+    }
 
     @Override
     public String serviceName() {