Implement Closeable

Author: lauzadis

runtime/runtime-core/native/src/aws/smithy/kotlin/runtime/compression/GzipCompressor.kt

@@ -4,13 +4,14 @@
  */
 package aws.smithy.kotlin.runtime.compression
 
+import aws.sdk.kotlin.crt.Closeable
 import kotlinx.cinterop.*
 import platform.zlib.*
 
 /**
  * Streaming-style gzip compressor, implemented using zlib bindings
  */
-internal class GzipCompressor {
+internal class GzipCompressor : Closeable {
     companion object {
         internal const val BUFFER_SIZE = 16384
     }
@@ -106,11 +107,15 @@ internal class GzipCompressor {
                 outputPin.unpin()
             }
 
-            deflateEnd(stream.ptr)
-            nativeHeap.free(stream.ptr)
-            isClosed = true
-
             return outputBuffer.toByteArray()
         }
     }
+
+    override fun close() {
+        if (isClosed) { return }
+
+        deflateEnd(stream.ptr)
+        nativeHeap.free(stream.ptr)
+        isClosed = true
+    }
 }

runtime/runtime-core/native/src/aws/smithy/kotlin/runtime/compression/GzipNative.kt

@@ -4,6 +4,7 @@
  */
 package aws.smithy.kotlin.runtime.compression
 
+import aws.sdk.kotlin.crt.use
 import aws.smithy.kotlin.runtime.content.ByteStream
 import aws.smithy.kotlin.runtime.io.GzipByteReadChannel
 import aws.smithy.kotlin.runtime.io.GzipSdkSource
@@ -38,9 +39,11 @@ public actual class Gzip : CompressionAlgorithm {
             if (sourceBytes.isEmpty()) {
                 stream
             } else {
-                val compressed = GzipCompressor().apply {
-                    update(sourceBytes)
-                }.flush()
+                val compressed = GzipCompressor().use {
+                    it.apply {
+                        update(sourceBytes)
+                    }.flush()
+                }
 
                 object : ByteStream.Buffer() {
                     override fun bytes(): ByteArray = compressed

runtime/runtime-core/native/src/aws/smithy/kotlin/runtime/io/GzipByteReadChannelNative.kt

@@ -47,7 +47,9 @@ public actual class GzipByteReadChannel actual constructor(public val channel: S
         if (compressor.availableForRead == 0 && channel.isClosedForRead) {
             val terminationBytes = compressor.flush()
             sink.write(terminationBytes)
-            return terminationBytes.size.toLong()
+            return terminationBytes.size.toLong().also {
+                compressor.close()
+            }
         }
 
         // Read compressed bytes from the compressor
@@ -58,7 +60,7 @@ public actual class GzipByteReadChannel actual constructor(public val channel: S
     }
 
     actual override fun cancel(cause: Throwable?): Boolean {
-        compressor.flush()
+        compressor.close()
         return channel.cancel(cause)
     }
 }

runtime/runtime-core/native/src/aws/smithy/kotlin/runtime/io/GzipSdkSourceNative.kt

@@ -35,7 +35,9 @@ public actual class GzipSdkSource actual constructor(public val source: SdkSourc
         if (compressor.availableForRead == 0) {
             val terminationBytes = compressor.flush()
             sink.write(terminationBytes)
-            return terminationBytes.size.toLong()
+            return terminationBytes.size.toLong().also {
+                compressor.close()
+            }
         }
 
         // Read compressed bytes from the compressor
@@ -46,7 +48,7 @@ public actual class GzipSdkSource actual constructor(public val source: SdkSourc
     }
 
     actual override fun close() {
-        compressor.flush()
+        compressor.close()
         source.close()
     }
 }