Updated aws-crt-client to use new 'utils' publishers for its request body logic.

Added two new publishers to 'utils':
1. StoringSubscriber - A subscriber that stores the events it receives in memory for retrieval.
2. ByteBufferStoringSubscriber - A subscriber that stores byte buffers it recevies in memory for retrieval.

Author: millems

build-tools/src/main/resources/software/amazon/awssdk/spotbugs-suppressions.xml

@@ -206,6 +206,13 @@
         <Bug pattern="URF_UNREAD_FIELD, RV_RETURN_VALUE_IGNORED"/>
     </Match>
 
+    <Match>
+        <Class name="software.amazon.awssdk.utils.async.StoringSubscriber"/>
+        <Method name="drop"/>
+        <Bug pattern="RV_RETURN_VALUE_IGNORED"/>
+    </Match>
+
+
     <Match>
         <Class name="software.amazon.awssdk.http.crt.internal.response.CrtResponseBodyPublisher" />
         <Method name="subscribe"/>

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

@@ -322,6 +322,8 @@ public interface Builder extends SdkAsyncHttpClient.Builder<AwsCrtAsyncHttpClien
          *
          * @param readBufferSize The number of bytes that can be buffered
          * @return The builder of the method chaining.
+         *
+         * TODO: This is also used for the write buffer size. Should we rename it?
          */
         Builder readBufferSize(int readBufferSize);
 

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

@@ -19,24 +19,27 @@
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.crt.http.HttpRequestBodyStream;
 import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
-import software.amazon.awssdk.utils.Validate;
+import software.amazon.awssdk.utils.async.ByteBufferStoringSubscriber;
+import software.amazon.awssdk.utils.async.ByteBufferStoringSubscriber.TransferResult;
 
-/**
- * Implements the CrtHttpStreamHandler API and converts CRT callbacks into calls to SDK AsyncExecuteRequest methods
- */
 @SdkInternalApi
 final class CrtRequestBodyAdapter implements HttpRequestBodyStream {
-    private final int windowSize;
-    private final CrtRequestBodySubscriber requestBodySubscriber;
+    private final SdkHttpContentPublisher requestPublisher;
+    private final ByteBufferStoringSubscriber requestBodySubscriber;
 
-    CrtRequestBodyAdapter(SdkHttpContentPublisher requestPublisher, int windowSize) {
-        this.windowSize = Validate.isPositive(windowSize, "windowSize is <= 0");
-        this.requestBodySubscriber = new CrtRequestBodySubscriber(windowSize);
+    CrtRequestBodyAdapter(SdkHttpContentPublisher requestPublisher, int readLimit) {
+        this.requestPublisher = requestPublisher;
+        this.requestBodySubscriber = new ByteBufferStoringSubscriber(readLimit);
         requestPublisher.subscribe(requestBodySubscriber);
     }
 
     @Override
     public boolean sendRequestBody(ByteBuffer bodyBytesOut) {
-        return requestBodySubscriber.transferRequestBody(bodyBytesOut);
+        return requestBodySubscriber.transferTo(bodyBytesOut) == TransferResult.END_OF_STREAM;
+    }
+
+    @Override
+    public long getLength() {
+        return requestPublisher.contentLength().orElse(0L);
     }
 }