Sync streaming compression

core/auth/src/main/java/software/amazon/awssdk/auth/signer/internal/chunkedencoding/AwsSignedChunkedEncodingInputStream.java

@@ -40,7 +40,6 @@
 @SdkInternalApi
 public final class AwsSignedChunkedEncodingInputStream extends AwsChunkedEncodingInputStream {
 
-    private static final String CRLF = "\r\n";
     private static final String CHUNK_SIGNATURE_HEADER = ";chunk-signature=";
     private static final String CHECKSUM_SIGNATURE_HEADER = "x-amz-trailer-signature:";
     private String previousChunkSignature;

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

@@ -17,44 +17,58 @@
 
 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.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.CompressRequestStage;
 
 /**
  * Interface for compressors used by {@link CompressRequestStage} to compress requests.
- * TODO: this will be refactored in the other PR
  */
-@SdkPublicApi
+@SdkInternalApi
 public interface Compressor {
 
     /**
      * The compression algorithm type.
+     *
+     * @return The {@link String} compression algorithm type.
      */
     String compressorType();
 
     /**
-     * Compress an {@link InputStream}.
+     * Compress a {@link SdkBytes} payload.
+     *
+     * @param content
+     * @return The compressed {@link SdkBytes}.
      */
-    InputStream compress(InputStream inputStream);
+    SdkBytes compress(SdkBytes content);
 
     /**
-     * Compress an async stream.
+     * Compress a byte[] payload.
+     *
+     * @param content
+     * @return The compressed byte array.
      */
-    Publisher<ByteBuffer> compressAsyncStream(Publisher<ByteBuffer> publisher);
+    default byte[] compress(byte[] content) {
+        return compress(SdkBytes.fromByteArray(content)).asByteArray();
+    }
+
+    /**
+     * Compress an {@link InputStream} payload.
+     *
+     * @param content
+     * @return The compressed {@link InputStream}.
+     */
+    default InputStream compress(InputStream content) {
+        return compress(SdkBytes.fromInputStream(content)).asInputStream();
+    }
 
     /**
-     * Maps the {@link CompressionType} to its corresponding {@link Compressor}.
-     * TODO: Update mappings here when additional compressors are supported in the future
+     * Compress an {@link ByteBuffer} payload.
+     *
+     * @param content
+     * @return The compressed {@link ByteBuffer}.
      */
-    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.");
-        }
+    default ByteBuffer compress(ByteBuffer content) {
+        return compress(SdkBytes.fromByteBuffer(content)).asByteBuffer();
     }
 }

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

@@ -0,0 +1,116 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.internal.compression.GzipCompressor;
+import software.amazon.awssdk.utils.Validate;
+
+/**
+ * The supported compression algorithms for operations with the requestCompression trait. Each supported algorithm will have an
+ * {@link Compressor} implementation.
+ */
+@SdkInternalApi
+public final class CompressorType {
+
+    public static final CompressorType GZIP = CompressorType.of("gzip");
+
+    private static Map<String, Compressor> compressorMap = new HashMap<String, Compressor>() {{
+            put("gzip", new GzipCompressor());
+        }};
+
+    private final String id;
+
+    private CompressorType(String id) {
+        this.id = id;
+    }
+
+    /**
+     * Creates a new {@link CompressorType} of the given value.
+     */
+    public static CompressorType of(String value) {
+        Validate.paramNotBlank(value, "compressionType");
+        return CompressorTypeCache.put(value);
+    }
+
+    /**
+     * Returns the {@link Set} of {@link String}s of compression types supported by the SDK.
+     */
+    public static Set<String> compressorTypes() {
+        return compressorMap.keySet();
+    }
+
+    /**
+     * Whether or not the compression type is supported by the SDK.
+     */
+    public static boolean isSupported(String compressionType) {
+        return compressorTypes().contains(compressionType);
+    }
+
+    /**
+     * Maps the {@link CompressorType} to its corresponding {@link Compressor}.
+     */
+    public Compressor newCompressor() {
+        Compressor compressor = compressorMap.getOrDefault(this.id, null);
+        if (compressor == null) {
+            throw new UnsupportedOperationException("The compression type " + id + " does not have an implementation of "
+                                                    + "Compressor");
+        }
+        return compressor;
+    }
+
+    @Override
+    public String toString() {
+        return id;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        CompressorType that = (CompressorType) o;
+        return Objects.equals(id, that.id)
+            && Objects.equals(compressorMap, that.compressorMap);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = id != null ? id.hashCode() : 0;
+        result = 31 * result + (compressorMap != null ? compressorMap.hashCode() : 0);
+        return result;
+    }
+
+    private static class CompressorTypeCache {
+        private static final ConcurrentHashMap<String, CompressorType> VALUES = new ConcurrentHashMap<>();
+
+        private CompressorTypeCache() {
+        }
+
+        private static CompressorType put(String value) {
+            return VALUES.computeIfAbsent(value, v -> new CompressorType(value));
+        }
+    }
+}

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

@@ -15,46 +15,42 @@
 
 package software.amazon.awssdk.core.internal.compression;
 
-import java.io.ByteArrayInputStream;
+import static software.amazon.awssdk.utils.IoUtils.closeQuietly;
+
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
+import java.io.UncheckedIOException;
 import java.util.zip.GZIPOutputStream;
-import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.core.SdkBytes;
 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";
+    private static final Logger log = LoggerFactory.getLogger(GzipCompressor.class);
 
     @Override
     public String compressorType() {
         return COMPRESSOR_TYPE;
     }
 
     @Override
-    public InputStream compress(InputStream inputStream) {
+    public SdkBytes compress(SdkBytes content) {
+        GZIPOutputStream gzipOutputStream = null;
         try {
-            byte[] content = IoUtils.toByteArray(inputStream);
             ByteArrayOutputStream compressedOutputStream = new ByteArrayOutputStream();
-            GZIPOutputStream gzipOutputStream = new GZIPOutputStream(compressedOutputStream);
-            gzipOutputStream.write(content);
+            gzipOutputStream = new GZIPOutputStream(compressedOutputStream);
+            gzipOutputStream.write(content.asByteArray());
             gzipOutputStream.close();
-
-            return new ByteArrayInputStream(compressedOutputStream.toByteArray());
+            return SdkBytes.fromByteArray(compressedOutputStream.toByteArray());
         } catch (IOException e) {
-            throw SdkClientException.create(e.getMessage(), e);
+            throw new UncheckedIOException(e);
+        } finally {
+            closeQuietly(gzipOutputStream, log);
         }
     }
-
-    @Override
-    public Publisher<ByteBuffer> compressAsyncStream(Publisher<ByteBuffer> publisher) {
-        //TODO
-        throw new UnsupportedOperationException();
-    }
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/AmazonAsyncHttpClient.java

@@ -37,6 +37,7 @@
 import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncExecutionFailureExceptionReportingStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncRetryableStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncSigningStage;
+import software.amazon.awssdk.core.internal.http.pipeline.stages.CompressRequestStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MakeRequestImmutableStage;
 import software.amazon.awssdk.core.internal.http.pipeline.stages.MakeRequestMutableStage;
@@ -169,6 +170,7 @@ public <OutputT> CompletableFuture<OutputT> execute(
                                 .then(ApplyUserAgentStage::new)
                                 .then(MergeCustomHeadersStage::new)
                                 .then(MergeCustomQueryParamsStage::new)
+                                .then(CompressRequestStage::new)
                                 .then(MakeRequestImmutableStage::new)
                                 .then(RequestPipelineBuilder
                                         .first(AsyncSigningStage::new)