Tck test for length subscriber

Author: dagnir

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/util/LengthCalculatingSubscriber.java

@@ -0,0 +1,56 @@
+/*
+ * 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.auth.aws.internal.signer.util;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+class LengthCalculatingSubscriber implements Subscriber<ByteBuffer> {
+    private final CompletableFuture<Long> contentLengthFuture = new CompletableFuture<>();
+    private Subscription subscription;
+    private long length = 0;
+
+    @Override
+    public void onSubscribe(Subscription subscription) {
+        if (this.subscription == null) {
+            this.subscription = subscription;
+            this.subscription.request(Long.MAX_VALUE);
+        } else {
+            subscription.cancel();
+        }
+    }
+
+    @Override
+    public void onNext(ByteBuffer byteBuffer) {
+        length += byteBuffer.remaining();
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        contentLengthFuture.completeExceptionally(throwable);
+    }
+
+    @Override
+    public void onComplete() {
+        contentLengthFuture.complete(length);
+    }
+
+    public CompletableFuture<Long> contentLengthFuture() {
+        return contentLengthFuture;
+    }
+}

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/util/SignerUtils.java

@@ -339,33 +339,4 @@ public static String getContentHash(SdkHttpRequest.Builder requestBuilder) {
     private static SdkChecksum sha256Checksum() {
         return SdkChecksum.forAlgorithm(() -> "SHA256");
     }
-
-    private static class LengthCalculatingSubscriber implements Subscriber<ByteBuffer> {
-        private final CompletableFuture<Long> contentLengthFuture = new CompletableFuture<>();
-        private long length = 0;
-
-        @Override
-        public void onSubscribe(Subscription subscription) {
-            subscription.request(Long.MAX_VALUE);
-        }
-
-        @Override
-        public void onNext(ByteBuffer byteBuffer) {
-            length += byteBuffer.remaining();
-        }
-
-        @Override
-        public void onError(Throwable throwable) {
-            contentLengthFuture.completeExceptionally(throwable);
-        }
-
-        @Override
-        public void onComplete() {
-            contentLengthFuture.complete(length);
-        }
-
-        public CompletableFuture<Long> contentLengthFuture() {
-            return contentLengthFuture;
-        }
-    }
 }

core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/internal/signer/util/LengthCalculatingSubscriberTckTest.java

@@ -0,0 +1,38 @@
+/*
+ * 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.auth.aws.internal.signer.util;
+
+import java.nio.ByteBuffer;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.tck.SubscriberBlackboxVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+public class LengthCalculatingSubscriberTckTest extends SubscriberBlackboxVerification<ByteBuffer> {
+
+    public LengthCalculatingSubscriberTckTest() {
+        super(new TestEnvironment());
+    }
+
+    @Override
+    public Subscriber<ByteBuffer> createSubscriber() {
+        return new LengthCalculatingSubscriber();
+    }
+
+    @Override
+    public ByteBuffer createElement(int element) {
+        return ByteBuffer.wrap(Integer.toString(element).getBytes());
+    }
+}