Updated crt http client to use latest crt api, which uses byte[] for response body.

Author: JonathanHenson

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.35</version>
+            <version>0.4.1</version>
         </dependency>
 
         <!--SDK dependencies-->

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

@@ -135,8 +135,7 @@ private HttpClientConnectionManager createConnectionPool(URI uri) {
         Validate.notNull(uri, "URI must not be null");
         log.debug(() -> "Creating ConnectionPool for: URI:" + uri + ", MaxConns: " + maxConnectionsPerEndpoint);
 
-        return new HttpClientConnectionManager(bootstrap, socketOptions, tlsContext, uri,
-                                                HttpClientConnectionManager.DEFAULT_MAX_BUFFER_SIZE, windowSize,
+        return new HttpClientConnectionManager(bootstrap, socketOptions, tlsContext, uri, windowSize,
                                                 maxConnectionsPerEndpoint);
     }
 

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

@@ -89,7 +89,7 @@ public void onResponseHeadersDone(HttpStream stream, int headerType) {
     }
 
     @Override
-    public int onResponseBody(HttpStream stream, ByteBuffer bodyBytesIn) {
+    public int onResponseBody(HttpStream stream, byte[] bodyBytesIn) {
         initRespBodyPublisherIfNeeded(stream);
 
         if (respBodyPublisher == null) {
@@ -99,10 +99,10 @@ public int onResponseBody(HttpStream stream, ByteBuffer bodyBytesIn) {
 
         // Queue a Deep Copy since bodyBytesIn is only guaranteed to contain valid memory for the lifetime of this
         // function call, and it's memory can be reused once this function returns.
-        respBodyPublisher.queueBuffer(deepCopy(bodyBytesIn));
+        respBodyPublisher.queueBuffer(bodyBytesIn);
         respBodyPublisher.publishToSubscribers();
 
-        if (bodyBytesIn.remaining() != 0) {
+        if (bodyBytesIn.length != 0) {
             throw new IllegalStateException("Unprocessed bytes remain in bodyBytesIn Buffer!");
         }
 

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

@@ -53,7 +53,7 @@ public class AwsCrtResponseBodyPublisher implements Publisher<ByteBuffer> {
     private final AtomicInteger mutualRecursionDepth = new AtomicInteger(0);
     private final AtomicInteger queuedBytes = new AtomicInteger(0);
     private final AtomicReference<Subscriber<? super ByteBuffer>> subscriberRef = new AtomicReference<>(null);
-    private final Queue<ByteBuffer> queuedBuffers = new ConcurrentLinkedQueue<>();
+    private final Queue<byte[]> queuedBuffers = new ConcurrentLinkedQueue<>();
     private final AtomicReference<Throwable> error = new AtomicReference<>(null);
 
     /**
@@ -97,18 +97,18 @@ public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
      * Adds a Buffer to the Queue to be published to any Subscribers
      * @param buffer The Buffer to be queued.
      */
-    public void queueBuffer(ByteBuffer buffer) {
+    public void queueBuffer(byte[] buffer) {
         Validate.notNull(buffer, "ByteBuffer must not be null");
 
         if (isCancelled.get()) {
             // Immediately open HttpStream's IO window so it doesn't see any IO Back-pressure.
             // AFAIK there's no way to abort an in-progress HttpStream, only free it's memory by calling close()
-            stream.incrementWindow(buffer.remaining());
+            stream.incrementWindow(buffer.length);
             return;
         }
 
         queuedBuffers.add(buffer);
-        int totalBytesQueued = queuedBytes.addAndGet(buffer.remaining());
+        int totalBytesQueued = queuedBytes.addAndGet(buffer.length);
 
         if (totalBytesQueued > windowSize) {
             throw new IllegalStateException("Queued more than Window Size: queued=" + totalBytesQueued
@@ -230,10 +230,10 @@ protected synchronized void publishToSubscribers() {
         int totalAmountTransferred = 0;
 
         while (outstandingRequests.get() > 0 && queuedBuffers.size() > 0) {
-            ByteBuffer buffer = queuedBuffers.poll();
+            byte[] buffer = queuedBuffers.poll();
             outstandingRequests.getAndUpdate(DECREMENT_IF_GREATER_THAN_ZERO);
-            int amount = buffer.remaining();
-            publishWithoutMutualRecursion(subscriberRef.get(), buffer);
+            int amount = buffer.length;
+            publishWithoutMutualRecursion(subscriberRef.get(), ByteBuffer.wrap(buffer));
             totalAmountTransferred += amount;
         }