Update CRT Client

Author: alexw91

http-clients/aws-crt-client/pom.xml

@@ -33,7 +33,7 @@
         <dependency>
             <groupId>software.amazon.awssdk.crt</groupId>
             <artifactId>aws-crt</artifactId>
-            <version>0.3.14</version>
+            <version>0.3.17</version>
         </dependency>
 
         <!--SDK dependencies-->

http-clients/aws-crt-client/src/it/java/software/amazon/awssdk/http/crt/AwsCrtClientS3IntegrationTest.java

@@ -20,12 +20,14 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.ResponseBytes;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.crt.CrtResource;
 import software.amazon.awssdk.crt.io.ClientBootstrap;
@@ -35,6 +37,7 @@
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 
 
 public class AwsCrtClientS3IntegrationTest {
@@ -43,9 +46,10 @@ public class AwsCrtClientS3IntegrationTest {
      */
     private static String BUCKET_NAME = "aws-crt-test-stuff";
 
-    private static String KEY = "http_test_doc.txt";
-
-    private static String FILE_SHA256 = "C7FDB5314B9742467B16BD5EA2F8012190B5E2C44A005F7984F89AAB58219534";
+    private static String LARGE_FILE = "http_test_doc.txt";
+    private static String SMALL_FILE = "random_32_byte.data";
+    private static String LARGE_FILE_SHA256 = "C7FDB5314B9742467B16BD5EA2F8012190B5E2C44A005F7984F89AAB58219534";
+    private static int NUM_REQUESTS = 1000;
 
     private static Region REGION = Region.US_EAST_1;
 
@@ -63,7 +67,7 @@ private void addResource(CrtResource resource) {
     public void setup() {
         Assert.assertEquals("Expected Zero allocated AwsCrtResources", 0, CrtResource.getAllocatedNativeResourceCount());
 
-        ClientBootstrap bootstrap = new ClientBootstrap(1);
+        ClientBootstrap bootstrap = new ClientBootstrap(4);
         SocketOptions socketOptions = new SocketOptions();
         TlsContext tlsContext = new TlsContext();
 
@@ -100,11 +104,31 @@ public void tearDown() {
     public void testDownloadFromS3() throws Exception {
         GetObjectRequest s3Request = GetObjectRequest.builder()
                 .bucket(BUCKET_NAME)
-                .key(KEY)
+                .key(LARGE_FILE)
                 .build();
 
         byte[] responseBody = s3.getObject(s3Request, AsyncResponseTransformer.toBytes()).get(120, TimeUnit.SECONDS).asByteArray();
 
-        assertThat(sha256Hex(responseBody).toUpperCase()).isEqualTo(FILE_SHA256);
+        assertThat(sha256Hex(responseBody).toUpperCase()).isEqualTo(LARGE_FILE_SHA256);
+    }
+
+    @Test
+    public void testParallelDownloadFromS3() throws Exception {
+        List<CompletableFuture<ResponseBytes<GetObjectResponse>> > requestFutures = new ArrayList<>();
+
+        for (int i = 0; i < NUM_REQUESTS; i++) {
+            GetObjectRequest s3Request = GetObjectRequest.builder()
+                    .bucket(BUCKET_NAME)
+                    .key(SMALL_FILE)
+                    .build();
+            CompletableFuture<ResponseBytes<GetObjectResponse>> requestFuture = s3.getObject(s3Request, AsyncResponseTransformer.toBytes());
+            requestFutures.add(requestFuture);
+        }
+
+        for(CompletableFuture<ResponseBytes<GetObjectResponse>>  f: requestFutures) {
+            f.join();
+            Assert.assertEquals(32, f.get().asByteArray().length);
+        }
     }
+
 }

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClient.java

@@ -57,6 +57,8 @@ public class AwsCrtAsyncHttpClient implements SdkAsyncHttpClient {
     private static final Logger log = Logger.loggerFor(AwsCrtAsyncHttpClient.class);
     private static final String HOST_HEADER = "Host";
     private static final String CONTENT_LENGTH = "Content-Length";
+    private static final String CONNECTION = "Connection";
+    private static final String KEEP_ALIVE = "keep-alive";
     private static final String AWS_COMMON_RUNTIME = "AwsCommonRuntime";
     private static final int DEFAULT_STREAM_WINDOW_SIZE = 16 * 1024 * 1024; // 16 MB Total Buffer size
     private static final int DEFAULT_HTTP_BODY_UPDATE_SIZE = 4 * 1024 * 1024; // 4 MB Update size from Native
@@ -111,7 +113,7 @@ public String clientName() {
 
     private HttpConnectionPoolManager createConnectionPool(URI uri) {
         Validate.notNull(uri, "URI must not be null");
-        log.debug(() -> "Creating ConnectionPool for: " + uri);
+        log.debug(() -> "Creating ConnectionPool for: URI:" + uri + ", MaxConns: " + maxConnectionsPerEndpoint);
         return new HttpConnectionPoolManager(bootstrap, socketOptions, tlsContext, uri, windowSize, maxConnectionsPerEndpoint);
     }
 
@@ -144,6 +146,11 @@ private List<HttpHeader> createHttpHeaderList(URI uri, AsyncExecuteRequest async
             crtHeaderList.add(new HttpHeader(HOST_HEADER, uri.getHost()));
         }
 
+        // Add Connection Keep Alive Header to reuse this Http Connection as long as possible
+        if (isNullOrEmpty(sdkRequest.headers().get(CONNECTION))) {
+            crtHeaderList.add(new HttpHeader(CONNECTION, KEEP_ALIVE));
+        }
+
         // Set Content-Length if needed
         Optional<Long> contentLength = asyncRequest.requestContentPublisher().contentLength();
         if (isNullOrEmpty(sdkRequest.headers().get(CONTENT_LENGTH)) && contentLength.isPresent()) {
@@ -200,19 +207,20 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
         reqOptions.setBodyBufferSize(httpBodyUpdateSize);
 
         // When a Connection is ready from the Connection Pool, schedule the Request on the connection
-        crtConnPool.acquireConnection().whenComplete((crtConn, throwable) -> {
-            // If we didn't get a connection for some reason, fail the request
-            if (throwable != null) {
-                requestFuture.completeExceptionally(throwable);
-                return;
-            }
-
-            // When the Request is complete, return our connection back to the Connection Pool
-            requestFuture.whenComplete((v, t) ->  crtConnPool.releaseConnection(crtConn));
-
-            // Submit the Request on this Connection
-            invokeSafely(() -> crtConn.makeRequest(crtRequest, reqOptions, crtToSdkAdapter));
-        });
+        crtConnPool.acquireConnection()
+            .whenComplete((crtConn, throwable) -> {
+                // If we didn't get a connection for some reason, fail the request
+                if (throwable != null) {
+                    requestFuture.completeExceptionally(throwable);
+                    return;
+                }
+
+                // When the Request is complete, return our connection back to the Connection Pool
+                requestFuture.whenComplete((v, t) ->  crtConnPool.releaseConnection(crtConn));
+
+                // Submit the Request on this Connection
+                invokeSafely(() -> crtConn.makeRequest(crtRequest, reqOptions, crtToSdkAdapter));
+            });
 
         return requestFuture;
     }

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtAsyncHttpStreamAdapter.java

@@ -37,20 +37,20 @@
 public class AwsCrtAsyncHttpStreamAdapter implements CrtHttpStreamHandler {
     private static final Logger log = Logger.loggerFor(AwsCrtAsyncHttpStreamAdapter.class);
     private final AsyncExecuteRequest sdkRequest;
-    private final CompletableFuture<Void> reqComplete;
+    private final CompletableFuture<Void> responseComplete;
     private final SdkHttpResponse.Builder respBuilder = SdkHttpResponse.builder();
     private final int windowSize;
     private final AwsCrtRequestBodySubscriber requestBodySubscriber;
     private AwsCrtResponseBodyPublisher respBodyPublisher = null;
 
-    public AwsCrtAsyncHttpStreamAdapter(CompletableFuture<Void> reqComplete, AsyncExecuteRequest sdkRequest,
+    public AwsCrtAsyncHttpStreamAdapter(CompletableFuture<Void> responseComplete, AsyncExecuteRequest sdkRequest,
                                         int windowSize) {
-        Validate.notNull(reqComplete, "reqComplete Future is null");
+        Validate.notNull(responseComplete, "reqComplete Future is null");
         Validate.notNull(sdkRequest, "AsyncExecuteRequest Future is null");
         Validate.isPositive(windowSize, "windowSize is <= 0");
 
         this.sdkRequest = sdkRequest;
-        this.reqComplete = reqComplete;
+        this.responseComplete = responseComplete;
         this.windowSize = windowSize;
         this.requestBodySubscriber = new AwsCrtRequestBodySubscriber(windowSize);
 
@@ -70,7 +70,7 @@ public void onResponseHeaders(HttpStream stream, int responseStatusCode, HttpHea
     public void onResponseHeadersDone(HttpStream stream, boolean hasBody) {
         respBuilder.statusCode(stream.getResponseStatusCode());
         sdkRequest.responseHandler().onHeaders(respBuilder.build());
-        respBodyPublisher = new AwsCrtResponseBodyPublisher(stream, windowSize);
+        respBodyPublisher = new AwsCrtResponseBodyPublisher(stream, responseComplete, windowSize);
 
 
         if (!hasBody) {
@@ -92,6 +92,10 @@ public int onResponseBody(HttpStream stream, ByteBuffer bodyBytesIn) {
         respBodyPublisher.queueBuffer(deepCopy(bodyBytesIn));
         respBodyPublisher.publishToSubscribers();
 
+        if (bodyBytesIn.remaining() != 0) {
+            throw new IllegalStateException("Unprocessed bytes remain in bodyBytesIn Buffer!");
+        }
+
         return 0;
     }
 
@@ -101,7 +105,7 @@ public void onResponseComplete(HttpStream stream, int errorCode) {
             log.debug(() -> "Response Completed Successfully");
             respBodyPublisher.setQueueComplete();
             respBodyPublisher.publishToSubscribers();
-            reqComplete.complete(null);
+            responseComplete.complete(null);
         } else {
             HttpException error = new HttpException(errorCode);
             log.error(() -> "Response Encountered an Error.", error);
@@ -115,7 +119,7 @@ public void onResponseComplete(HttpStream stream, int errorCode) {
                 respBodyPublisher.publishToSubscribers();
             }
 
-            reqComplete.completeExceptionally(error);
+            responseComplete.completeExceptionally(error);
         }
 
         stream.close();

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtRequestBodySubscriber.java

@@ -126,7 +126,7 @@ public synchronized boolean transferRequestBody(ByteBuffer out) {
         if (!endOfStream) {
             requestDataIfNecessary();
         } else {
-            log.debug(() -> "End Of Stream reached");
+            log.debug(() -> "End Of RequestBody reached");
         }
 
         return endOfStream;

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/AwsCrtResponseBodyPublisher.java

@@ -17,6 +17,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -38,6 +39,7 @@ public class AwsCrtResponseBodyPublisher implements Publisher<ByteBuffer> {
     private static final Logger log = Logger.loggerFor(AwsCrtResponseBodyPublisher.class);
     private static final LongUnaryOperator DECREMENT_IF_GREATER_THAN_ZERO = x -> ((x > 0) ? (x - 1) : (x));
 
+    private final CompletableFuture<Void> responseComplete;
     private final AtomicLong outstandingRequests = new AtomicLong(0);
     private final HttpStream stream;
     private final int windowSize;
@@ -56,10 +58,12 @@ public class AwsCrtResponseBodyPublisher implements Publisher<ByteBuffer> {
      * @param windowSize The max allowed bytes to be queued. The sum of the sizes of all queued ByteBuffers should
      *                   never exceed this value.
      */
-    public AwsCrtResponseBodyPublisher(HttpStream stream, int windowSize) {
+    public AwsCrtResponseBodyPublisher(HttpStream stream, CompletableFuture<Void> responseComplete, int windowSize) {
         Validate.notNull(stream, "Stream must not be null");
+        Validate.notNull(responseComplete, "Stream must not be null");
         Validate.isPositive(windowSize, "windowSize must be > 0");
         this.stream = stream;
+        this.responseComplete = responseComplete;
         this.windowSize = windowSize;
     }
 
@@ -153,18 +157,21 @@ protected void completeSubscriptionExactlyOnce() {
             return;
         }
 
-        Subscriber s = subscriberRef.getAndSet(null);
+        Subscriber subscriber = subscriberRef.getAndSet(null);
 
-        if (s == null) {
+        if (subscriber == null) {
             return;
         }
 
         Throwable throwable = error.get();
 
         if (throwable != null) {
-            s.onError(throwable);
+            log.error(() -> "Error before ResponseBodyPublisher could complete: " + throwable.getMessage());
+            subscriber.onError(throwable);
         } else {
-            s.onComplete();
+            log.debug(() -> "ResponseBodyPublisher Completed Successfully");
+            subscriber.onComplete();
+            responseComplete.complete(null);
         }
     }
 

http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/AwsCrtResponseBodyPublisherReactiveStreamCompatTest.java

@@ -19,6 +19,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.tck.PublisherVerification;
 import org.reactivestreams.tck.TestEnvironment;
@@ -36,7 +37,7 @@ public AwsCrtResponseBodyPublisherReactiveStreamCompatTest() {
     @Override
     public Publisher<ByteBuffer> createPublisher(long elements) {
         HttpStream stream = mock(HttpStream.class);
-        AwsCrtResponseBodyPublisher bodyPublisher = new AwsCrtResponseBodyPublisher(stream, Integer.MAX_VALUE);
+        AwsCrtResponseBodyPublisher bodyPublisher = new AwsCrtResponseBodyPublisher(stream, new CompletableFuture<>(), Integer.MAX_VALUE);
 
         for (long i = 0; i < elements; i++) {
             bodyPublisher.queueBuffer(ByteBuffer.wrap(UUID.randomUUID().toString().getBytes()));