refactor(rt)!: update SDK I/O abstractions (#751)

Author: aajtodd

.changes/54b5e8b9-394c-4235-b66c-6b912fdb106b.json

@@ -0,0 +1,5 @@
+{
+    "id": "54b5e8b9-394c-4235-b66c-6b912fdb106b",
+    "type": "misc",
+    "description": "**BREAKING** Refactor SDK I/O types. See [this discussion post](https://github.com/awslabs/aws-sdk-kotlin/discussions/768) for more information"
+}

docs/design/binary-streaming.md

@@ -6,7 +6,7 @@ kotlin.native.ignoreDisabledTargets=true
 kotlin.mpp.enableCompatibilityMetadataVariant=true
 
 # SDK
-sdkVersion=0.13.2-SNAPSHOT
+sdkVersion=0.14.0-SNAPSHOT
 
 # kotlin
 kotlinVersion=1.7.10
@@ -19,6 +19,7 @@ atomicFuVersion=0.18.3
 kotlinxSerializationVersion=1.3.3
 jsoupVersion=1.14.3
 okHttpVersion=5.0.0-alpha.10
+okioVersion=3.2.0
 
 # codegen
 smithyVersion=1.26.1
@@ -43,4 +44,4 @@ kotlinLoggingVersion=2.1.21
 slf4jVersion=1.7.36
 
 # crt
-crtKotlinVersion=0.6.6
+crtKotlinVersion=0.6.6

gradle.properties

@@ -5,7 +5,9 @@
 
 package aws.smithy.kotlin.runtime.auth.awssigning
 
+import aws.smithy.kotlin.runtime.auth.awssigning.internal.AwsChunkedReader
 import aws.smithy.kotlin.runtime.http.Headers
+import aws.smithy.kotlin.runtime.io.SdkBuffer
 import aws.smithy.kotlin.runtime.io.SdkByteReadChannel
 import aws.smithy.kotlin.runtime.util.InternalApi
 
@@ -14,17 +16,48 @@ import aws.smithy.kotlin.runtime.util.InternalApi
  * This class wraps an SdkByteReadChannel. When reads are performed on this class, it will read the wrapped data
  * and return it in aws-chunked content encoding.
  * @see <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html">SigV4 Streaming</a>
- * @param chan the underlying [SdkByteReadChannel] which will have its data encoded in aws-chunked format
+ * @param delegate the underlying [SdkByteReadChannel] which will have its data encoded in aws-chunked format
  * @param signer the signer to use to sign chunks and (optionally) chunk trailer
  * @param signingConfig the config to use for signing
  * @param previousSignature the previous signature to use for signing. in most cases, this should be the seed signature
  * @param trailingHeaders the optional trailing headers to include in the final chunk
  */
 @InternalApi
-public expect class AwsChunkedByteReadChannel public constructor(
-    chan: SdkByteReadChannel,
-    signer: AwsSigner,
-    signingConfig: AwsSigningConfig,
-    previousSignature: ByteArray,
-    trailingHeaders: Headers = Headers.Empty,
-) : AbstractAwsChunkedByteReadChannel
+public class AwsChunkedByteReadChannel(
+    private val delegate: SdkByteReadChannel,
+    private val signer: AwsSigner,
+    private val signingConfig: AwsSigningConfig,
+    private var previousSignature: ByteArray,
+    private val trailingHeaders: Headers = Headers.Empty,
+) : SdkByteReadChannel by delegate {
+
+    private val chunkReader = AwsChunkedReader(
+        delegate.asStream(),
+        signer,
+        signingConfig,
+        previousSignature,
+        trailingHeaders,
+    )
+
+    override val isClosedForRead: Boolean
+        get() = chunkReader.chunk.size == 0L && chunkReader.hasLastChunkBeenSent && delegate.isClosedForRead
+
+    override val availableForRead: Int
+        get() = chunkReader.chunk.size.toInt() + delegate.availableForRead
+
+    override suspend fun read(sink: SdkBuffer, limit: Long): Long {
+        require(limit >= 0L) { "Invalid limit ($limit) must be >= 0L" }
+        if (!chunkReader.ensureValidChunk()) return -1L
+        return chunkReader.chunk.read(sink, limit)
+    }
+}
+
+private fun SdkByteReadChannel.asStream(): AwsChunkedReader.Stream = object : AwsChunkedReader.Stream {
+    private val delegate = this@asStream
+
+    override fun isClosedForRead(): Boolean =
+        delegate.isClosedForRead
+
+    override suspend fun read(sink: SdkBuffer, limit: Long): Long =
+        delegate.read(sink, limit)
+}