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/CompressionType.java

@@ -15,60 +15,80 @@
 
 package software.amazon.awssdk.core.compression;
 
-import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.utils.internal.EnumUtils;
+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 enum CompressionType {
+public final class CompressionType {
 
-    GZIP("gzip"),
+    public static final CompressionType GZIP = CompressionType.of("gzip");
 
-    UNKNOWN_TO_SDK_VERSION(null);
+    private static Map<String, Compressor> compressorMap = new HashMap<String, Compressor>() {{
+            put("gzip", new GzipCompressor());
+        }};
 
-    private static final Map<String, CompressionType> VALUE_MAP = EnumUtils.uniqueIndex(
-        CompressionType.class, CompressionType::toString);
+    private String id;
 
-    private final String value;
+    private CompressionType(String id) {
+        this.id = id;
+    }
 
-    CompressionType(String value) {
-        this.value = value;
+    /**
+     * Creates a new {@link CompressionType} of the given value.
+     */
+    public static CompressionType of(String value) {
+        Validate.paramNotBlank(value, "compressionType");
+        return CompressionTypeCache.put(value);
     }
 
-    @Override
-    public String toString() {
-        return String.valueOf(value);
+    /**
+     * Returns the {@link Set} of {@link String}s of compression types supported by the SDK.
+     */
+    public static Set<String> compressionTypes() {
+        return compressorMap.keySet();
     }
 
     /**
-     * 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
+     * Whether or not the compression type is supported by the SDK.
      */
-    public static CompressionType fromValue(String value) {
-        if (value == null) {
-            return null;
-        }
-        return VALUE_MAP.getOrDefault(value, UNKNOWN_TO_SDK_VERSION);
+    public static boolean isSupported(String compressionType) {
+        return compressionTypes().contains(compressionType);
     }
 
     /**
-     * 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
+     * Maps the {@link CompressionType} to its corresponding {@link Compressor}.
      */
-    public static Set<CompressionType> knownValues() {
-        Set<CompressionType> knownValues = EnumSet.allOf(CompressionType.class);
-        knownValues.remove(UNKNOWN_TO_SDK_VERSION);
-        return knownValues;
+    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;
+    }
+
+    private static class CompressionTypeCache {
+        private static final ConcurrentHashMap<String, CompressionType> VALUES = new ConcurrentHashMap<>();
+
+        private CompressionTypeCache() {
+        }
+
+        private static CompressionType put(String value) {
+            return VALUES.computeIfAbsent(value, v -> new CompressionType(value));
+        }
     }
 }

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/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)