Request compression interceptor and non-streaming compression (#4124)

* Request compression interceptor and non-streaming compression

* Refactoring

* Refactoring

* Refactoring and implement sync streaming compression

* Refactoring

* Refactoring

* Refactoring and remove sync compression updates

Author: davidh44

core/sdk-core/src/main/java/software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java

@@ -77,6 +77,7 @@
 import software.amazon.awssdk.core.internal.interceptor.HttpChecksumInHeaderInterceptor;
 import software.amazon.awssdk.core.internal.interceptor.HttpChecksumRequiredInterceptor;
 import software.amazon.awssdk.core.internal.interceptor.HttpChecksumValidationInterceptor;
+import software.amazon.awssdk.core.internal.interceptor.RequestCompressionInterceptor;
 import software.amazon.awssdk.core.internal.interceptor.SyncHttpChecksumInTrailerInterceptor;
 import software.amazon.awssdk.core.retry.RetryMode;
 import software.amazon.awssdk.core.retry.RetryPolicy;
@@ -444,6 +445,7 @@ private List<ExecutionInterceptor> resolveExecutionInterceptors(SdkClientConfigu
      */
     private List<ExecutionInterceptor> sdkInterceptors() {
         return Collections.unmodifiableList(Arrays.asList(
+            new RequestCompressionInterceptor(),
             new HttpChecksumRequiredInterceptor(),
             new SyncHttpChecksumInTrailerInterceptor(),
             new HttpChecksumValidationInterceptor(),

core/sdk-core/src/main/java/software/amazon/awssdk/core/compression/CompressionType.java

@@ -0,0 +1,74 @@
+/*
+ * 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.core.compression;
+
+import java.util.EnumSet;
+import java.util.Map;
+import java.util.Set;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.utils.internal.EnumUtils;
+
+/**
+ * The supported compression algorithms for operations with the requestCompression trait. Each supported algorithm will have an
+ * {@link Compressor} implementation.
+ */
+@SdkInternalApi
+public enum CompressionType {
+
+    GZIP("gzip"),
+
+    UNKNOWN_TO_SDK_VERSION(null);
+
+    private static final Map<String, CompressionType> VALUE_MAP = EnumUtils.uniqueIndex(
+        CompressionType.class, CompressionType::toString);
+
+    private final String value;
+
+    CompressionType(String value) {
+        this.value = value;
+    }
+
+    @Override
+    public String toString() {
+        return String.valueOf(value);
+    }
+
+    /**
+     * Use this in place of valueOf to convert the raw string into the enum value.
+     *
+     * @param value
+     *        real value
+     * @return SupportedEncodings corresponding to the value
+     */
+    public static CompressionType fromValue(String value) {
+        if (value == null) {
+            return null;
+        }
+        return VALUE_MAP.getOrDefault(value, UNKNOWN_TO_SDK_VERSION);
+    }
+
+    /**
+     * Use this in place of {@link #values()} to return a {@link Set} of all values known to the SDK. This will return
+     * all known enum values except {@link #UNKNOWN_TO_SDK_VERSION}.
+     *
+     * @return a {@link Set} of known {@link CompressionType}s
+     */
+    public static Set<CompressionType> knownValues() {
+        Set<CompressionType> knownValues = EnumSet.allOf(CompressionType.class);
+        knownValues.remove(UNKNOWN_TO_SDK_VERSION);
+        return knownValues;
+    }
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/compression/Compressor.java

@@ -0,0 +1,59 @@
+/*
+ * 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.core.compression;
+
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import org.reactivestreams.Publisher;
+import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.core.internal.compression.GzipCompressor;
+import software.amazon.awssdk.core.internal.interceptor.RequestCompressionInterceptor;
+
+/**
+ * Interface for compressors to be used by {@link RequestCompressionInterceptor} to compress requests.
+ */
+@SdkPublicApi
+public interface Compressor {
+
+    /**
+     * The compression algorithm type.
+     */
+    String compressorType();
+
+    /**
+     * Compress an {@link InputStream}.
+     */
+    InputStream compress(InputStream inputStream);
+
+    /**
+     * Compress an async stream.
+     */
+    Publisher<ByteBuffer> compressAsyncStream(Publisher<ByteBuffer> publisher);
+
+    /**
+     * Maps the {@link CompressionType} to its corresponding {@link Compressor}.
+     * TODO: Update mappings here when additional compressors are supported in the future
+     */
+    static Compressor forCompressorType(CompressionType compressionType) {
+        switch (compressionType) {
+            case GZIP:
+                return new GzipCompressor();
+            default:
+                throw new IllegalArgumentException("The compresssion type " + compressionType + "does not have an implemenation"
+                                                   + " of Compressor.");
+        }
+    }
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/compression/GzipCompressor.java

@@ -0,0 +1,60 @@
+/*
+ * 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.core.internal.compression;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.zip.GZIPOutputStream;
+import org.reactivestreams.Publisher;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.compression.Compressor;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.utils.IoUtils;
+
+@SdkInternalApi
+public final class GzipCompressor implements Compressor {
+
+    private static final String COMPRESSOR_TYPE = "gzip";
+
+    @Override
+    public String compressorType() {
+        return COMPRESSOR_TYPE;
+    }
+
+    @Override
+    public InputStream compress(InputStream inputStream) {
+        try {
+            byte[] content = IoUtils.toByteArray(inputStream);
+            ByteArrayOutputStream compressedOutputStream = new ByteArrayOutputStream();
+            GZIPOutputStream gzipOutputStream = new GZIPOutputStream(compressedOutputStream);
+            gzipOutputStream.write(content);
+            gzipOutputStream.close();
+
+            return new ByteArrayInputStream(compressedOutputStream.toByteArray());
+        } catch (IOException e) {
+            throw SdkClientException.create(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public Publisher<ByteBuffer> compressAsyncStream(Publisher<ByteBuffer> publisher) {
+        //TODO
+        throw new UnsupportedOperationException();
+    }
+}