Add flexible checksum support for non-streaming cases (#4376)

* Add flexible checksum support for non-streaming cases

* Add more tests, address comments

* Get rid of reflection, hasTrailer(), and address other comments

* Remove DefaultChecksummer, update namings

* Address last comments

core/http-auth-aws/pom.xml

@@ -59,6 +59,22 @@
             <artifactId>http-auth-spi</artifactId>
             <version>${awsjavasdk.version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>checksums-spi</artifactId>
+            <version>${awsjavasdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>checksums</artifactId>
+            <version>${awsjavasdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk.crt</groupId>
+            <artifactId>aws-crt</artifactId>
+            <version>${awscrt.version}</version>
+            <optional>true</optional>
+        </dependency>
 
         <dependency>
             <groupId>org.mockito</groupId>
@@ -85,6 +101,11 @@
             <artifactId>equalsverifier</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams-tck</artifactId>
+            <scope>test</scope>
+        </dependency>
 
     </dependencies>
 

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/AwsV4HttpSigner.java

@@ -17,6 +17,7 @@
 
 import java.time.Duration;
 import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.checksums.spi.ChecksumAlgorithm;
 import software.amazon.awssdk.http.auth.aws.internal.signer.DefaultAwsV4HttpSigner;
 import software.amazon.awssdk.http.auth.spi.HttpSigner;
 import software.amazon.awssdk.http.auth.spi.SignerProperty;
@@ -83,6 +84,12 @@ public interface AwsV4HttpSigner extends HttpSigner<AwsCredentialsIdentity> {
     SignerProperty<Boolean> CHUNK_ENCODING_ENABLED =
         SignerProperty.create(Boolean.class, "ChunkEncodingEnabled");
 
+    /**
+     * The algorithm to use for calculating a "flexible" checksum. This property is optional.
+     */
+    SignerProperty<ChecksumAlgorithm> CHECKSUM_ALGORITHM =
+        SignerProperty.create(ChecksumAlgorithm.class, "ChecksumAlgorithm");
+
 
     /**
      * Get a default implementation of a {@link AwsV4HttpSigner}

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/checksums/ConstantChecksum.java

@@ -0,0 +1,63 @@
+/*
+ * 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.checksums;
+
+import java.nio.charset.StandardCharsets;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+
+/**
+ * Implementation of {@link SdkChecksum} to provide a constant checksum.
+ */
+@SdkInternalApi
+public class ConstantChecksum implements SdkChecksum {
+
+    private final String value;
+
+    public ConstantChecksum(String value) {
+        this.value = value;
+    }
+
+    @Override
+    public void update(int b) {
+    }
+
+    @Override
+    public void update(byte[] b, int off, int len) {
+    }
+
+    @Override
+    public long getValue() {
+        throw new UnsupportedOperationException("Use getChecksumBytes() instead.");
+    }
+
+    @Override
+    public void reset() {
+    }
+
+    @Override
+    public byte[] getChecksumBytes() {
+        return value.getBytes(StandardCharsets.UTF_8);
+    }
+
+    @Override
+    public void mark(int readLimit) {
+    }
+
+    @Override
+    public String getChecksum() {
+        return value;
+    }
+}

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/checksums/Crc32CChecksum.java

@@ -0,0 +1,110 @@
+/*
+ * 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.checksums;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.zip.Checksum;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.crt.checksums.CRC32C;
+import software.amazon.awssdk.utils.ClassLoaderHelper;
+
+/**
+ * Implementation of {@link SdkChecksum} to calculate an CRC32C checksum.
+ */
+@SdkInternalApi
+public class Crc32CChecksum implements SdkChecksum {
+
+    private static final String CRT_CLASSPATH_FOR_CRC32C = "software.amazon.awssdk.crt.checksums.CRC32C";
+
+    private Checksum crc32c;
+    private Checksum lastMarkedCrc32C;
+
+    /**
+     * Creates CRT Based Crc32C checksum if Crt classpath for Crc32c is loaded, else create Sdk Implemented Crc32c
+     */
+    public Crc32CChecksum() {
+        if (isCrtAvailable()) {
+            crc32c = new CRC32C();
+        } else {
+            crc32c = SdkCrc32CChecksum.create();
+        }
+    }
+
+    @Override
+    public byte[] getChecksumBytes() {
+        return Arrays.copyOfRange(longToByte(crc32c.getValue()), 4, 8);
+    }
+
+
+    @Override
+    public void mark(int readLimit) {
+        this.lastMarkedCrc32C = cloneChecksum(crc32c);
+    }
+
+    @Override
+    public void update(int b) {
+        crc32c.update(b);
+    }
+
+    @Override
+    public void update(byte[] b, int off, int len) {
+        crc32c.update(b, off, len);
+    }
+
+    @Override
+    public long getValue() {
+        return crc32c.getValue();
+    }
+
+    @Override
+    public void reset() {
+        if (lastMarkedCrc32C == null) {
+            crc32c.reset();
+        } else {
+            crc32c = cloneChecksum(lastMarkedCrc32C);
+        }
+    }
+
+
+    private Checksum cloneChecksum(Checksum checksum) {
+        if (checksum instanceof CRC32C) {
+            return (Checksum) ((CRC32C) checksum).clone();
+        }
+
+        if (checksum instanceof SdkCrc32CChecksum) {
+            return (Checksum) ((SdkCrc32CChecksum) checksum).clone();
+        }
+
+        throw new IllegalStateException("Unsupported checksum");
+    }
+
+    public static boolean isCrtAvailable() {
+        try {
+            ClassLoaderHelper.loadClass(CRT_CLASSPATH_FOR_CRC32C, false);
+        } catch (ClassNotFoundException e) {
+            return false;
+        }
+
+        return true;
+    }
+
+    private static byte[] longToByte(Long input) {
+        ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+        buffer.putLong(input);
+        return buffer.array();
+    }
+}

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/checksums/Crc32Checksum.java

@@ -0,0 +1,108 @@
+/*
+ * 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.checksums;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.zip.Checksum;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.crt.checksums.CRC32;
+import software.amazon.awssdk.utils.ClassLoaderHelper;
+
+/**
+ * Implementation of {@link SdkChecksum} to calculate an CRC32 checksum.
+ */
+@SdkInternalApi
+public class Crc32Checksum implements SdkChecksum {
+
+    private static final String CRT_CLASSPATH_FOR_CRC32 = "software.amazon.awssdk.crt.checksums.CRC32";
+
+    private Checksum crc32;
+    private Checksum lastMarkedCrc32;
+
+    /**
+     * Creates CRT Based Crc32 checksum if Crt classpath for Crc32 is loaded, else create Sdk Implemented Crc32.
+     */
+    public Crc32Checksum() {
+        if (isCrtAvailable()) {
+            crc32 = new CRC32();
+        } else {
+            crc32 = SdkCrc32Checksum.create();
+        }
+    }
+
+    @Override
+    public byte[] getChecksumBytes() {
+        return Arrays.copyOfRange(longToByte(crc32.getValue()), 4, 8);
+    }
+
+    @Override
+    public void mark(int readLimit) {
+        this.lastMarkedCrc32 = cloneChecksum(crc32);
+    }
+
+    @Override
+    public void update(int b) {
+        crc32.update(b);
+    }
+
+    @Override
+    public void update(byte[] b, int off, int len) {
+        crc32.update(b, off, len);
+    }
+
+    @Override
+    public long getValue() {
+        return crc32.getValue();
+    }
+
+    @Override
+    public void reset() {
+        if ((lastMarkedCrc32 == null)) {
+            crc32.reset();
+        } else {
+            crc32 = cloneChecksum(lastMarkedCrc32);
+        }
+    }
+
+    private Checksum cloneChecksum(Checksum checksum) {
+        if (checksum instanceof CRC32) {
+            return (Checksum) ((CRC32) checksum).clone();
+        }
+
+        if (checksum instanceof SdkCrc32Checksum) {
+            return (Checksum) ((SdkCrc32Checksum) checksum).clone();
+        }
+
+        throw new IllegalStateException("Unsupported checksum");
+    }
+
+    public static boolean isCrtAvailable() {
+        try {
+            ClassLoaderHelper.loadClass(CRT_CLASSPATH_FOR_CRC32, false);
+        } catch (ClassNotFoundException e) {
+            return false;
+        }
+
+        return true;
+    }
+
+    private static byte[] longToByte(Long input) {
+        ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+        buffer.putLong(input);
+        return buffer.array();
+    }
+}