Revert "Move CrtRequestBodyAdapter to crt-core"

This reverts commit f5876e0.

Author: dagnir

core/http-auth-aws/pom.xml

@@ -69,12 +69,6 @@
             <artifactId>checksums</artifactId>
             <version>${awsjavasdk.version}</version>
         </dependency>
-        <dependency>
-            <groupId>software.amazon.awssdk</groupId>
-            <artifactId>crt-core</artifactId>
-            <version>${awsjavasdk.version}</version>
-            <optional>true</optional>
-        </dependency>
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>http-auth-aws-crt</artifactId>

…wssdk/crtcore/CrtRequestBodyAdapter.java …ternal/signer/CrtRequestBodyAdapter.javacore/crt-core/src/main/java/software/amazon/awssdk/crtcore/CrtRequestBodyAdapter.java renamed to core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/crt/internal/signer/CrtRequestBodyAdapter.java core/crt-core/src/main/java/software/amazon/awssdk/crtcore/CrtRequestBodyAdapter.java renamed to core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/crt/internal/signer/CrtRequestBodyAdapter.java

@@ -13,7 +13,7 @@
  * permissions and limitations under the License.
  */
 
-package software.amazon.awssdk.crtcore;
+package software.amazon.awssdk.http.auth.aws.crt.internal.signer;
 
 import java.nio.ByteBuffer;
 import org.reactivestreams.Publisher;
@@ -29,14 +29,10 @@ public final class CrtRequestBodyAdapter implements HttpRequestBodyStream {
     private final long contentLength;
     private ByteBufferStoringSubscriber requestBodySubscriber;
 
-    public CrtRequestBodyAdapter(Publisher<ByteBuffer> requestPublisher, long contentLength, long readLimit) {
+    public CrtRequestBodyAdapter(Publisher<ByteBuffer> requestPublisher, long contentLength) {
         this.requestPublisher = requestPublisher;
         this.contentLength = contentLength;
-        this.requestBodySubscriber = new ByteBufferStoringSubscriber(readLimit);
-    }
-
-    public CrtRequestBodyAdapter(Publisher<ByteBuffer> requestPublisher, long contentLength) {
-        this(requestPublisher, contentLength, BUFFER_SIZE);
+        this.requestBodySubscriber = new ByteBufferStoringSubscriber(BUFFER_SIZE);
     }
 
     @Override

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/crt/internal/util/CrtHttpRequestConverter.java

@@ -28,10 +28,10 @@
 import software.amazon.awssdk.crt.http.HttpHeader;
 import software.amazon.awssdk.crt.http.HttpRequest;
 import software.amazon.awssdk.crt.http.HttpRequestBodyStream;
-import software.amazon.awssdk.crtcore.CrtRequestBodyAdapter;
 import software.amazon.awssdk.http.ContentStreamProvider;
 import software.amazon.awssdk.http.SdkHttpRequest;
 import software.amazon.awssdk.http.auth.aws.crt.internal.io.CrtInputStream;
+import software.amazon.awssdk.http.auth.aws.crt.internal.signer.CrtRequestBodyAdapter;
 import software.amazon.awssdk.utils.StringUtils;
 import software.amazon.awssdk.utils.http.SdkHttpUtils;
 import software.amazon.awssdk.utils.uri.SdkUri;

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

@@ -22,12 +22,10 @@
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.crt.http.HttpHeader;
 import software.amazon.awssdk.crt.http.HttpRequest;
-import software.amazon.awssdk.crtcore.CrtRequestBodyAdapter;
 import software.amazon.awssdk.http.Header;
 import software.amazon.awssdk.http.HttpExecuteRequest;
 import software.amazon.awssdk.http.SdkHttpRequest;
 import software.amazon.awssdk.http.async.AsyncExecuteRequest;
-import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
 import software.amazon.awssdk.http.crt.internal.CrtAsyncRequestContext;
 import software.amazon.awssdk.http.crt.internal.CrtRequestContext;
 
@@ -52,13 +50,10 @@ public static HttpRequest toAsyncCrtRequest(CrtAsyncRequestContext request) {
 
         HttpHeader[] crtHeaderArray = asArray(createAsyncHttpHeaderList(sdkRequest.getUri(), sdkExecuteRequest));
 
-        SdkHttpContentPublisher contentPublisher = sdkExecuteRequest.requestContentPublisher();
-
         return new HttpRequest(method,
                                encodedPath + encodedQueryString,
                                crtHeaderArray,
-                               new CrtRequestBodyAdapter(contentPublisher,
-                                                         contentPublisher.contentLength().orElse(0L),
+                               new CrtRequestBodyAdapter(sdkExecuteRequest.requestContentPublisher(),
                                                          request.readBufferSize()));
     }
 

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

@@ -0,0 +1,50 @@
+/*
+ * 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.http.crt.internal.request;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicBoolean;
+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.async.ByteBufferStoringSubscriber;
+import software.amazon.awssdk.utils.async.ByteBufferStoringSubscriber.TransferResult;
+
+@SdkInternalApi
+final class CrtRequestBodyAdapter implements HttpRequestBodyStream {
+    private final SdkHttpContentPublisher requestPublisher;
+    private final ByteBufferStoringSubscriber requestBodySubscriber;
+    private final AtomicBoolean subscribed = new AtomicBoolean(false);
+
+    CrtRequestBodyAdapter(SdkHttpContentPublisher requestPublisher, long readLimit) {
+        this.requestPublisher = requestPublisher;
+        this.requestBodySubscriber = new ByteBufferStoringSubscriber(readLimit);
+    }
+
+    @Override
+    public boolean sendRequestBody(ByteBuffer bodyBytesOut) {
+        if (subscribed.compareAndSet(false, true)) {
+            requestPublisher.subscribe(requestBodySubscriber);
+        }
+
+        return requestBodySubscriber.transferTo(bodyBytesOut) == TransferResult.END_OF_STREAM;
+    }
+
+    @Override
+    public long getLength() {
+        return requestPublisher.contentLength().orElse(0L);
+    }
+}