fix: readAvailable fallback behavior (#620)

Author: aajtodd

runtime/io/common/src/aws/smithy/kotlin/runtime/io/SdkByteReadChannel.kt

@@ -158,6 +158,7 @@ internal suspend fun SdkByteReadChannel.readAvailableFallback(dest: SdkByteBuffe
     // channel was closed while waiting and no further content was made available
     if (availableForRead == 0 && isClosedForRead) return -1
     val tmp = ByteArray(minOf(availableForRead.toLong(), limit, Int.MAX_VALUE.toLong()).toInt())
+    readFully(tmp)
     dest.writeFully(tmp)
     return tmp.size.toLong()
 }

runtime/io/common/test/aws/smithy/kotlin/runtime/io/SdkByteChannelOpsTest.kt

@@ -170,4 +170,36 @@ class SdkByteChannelOpsTest {
         }
         assertNull(chan.readUtf8CodePoint())
     }
+
+    private class ProxyChan(
+        val ch: SdkByteReadChannel
+    ) : SdkByteReadChannel by ch {
+        var proxyCalled = false
+        override suspend fun readRemaining(limit: Int): ByteArray {
+            proxyCalled = true
+            return ch.readRemaining(limit)
+        }
+        override suspend fun readFully(sink: ByteArray, offset: Int, length: Int) {
+            proxyCalled = true
+            return ch.readFully(sink, offset, length)
+        }
+        override suspend fun readAvailable(sink: ByteArray, offset: Int, length: Int): Int {
+            proxyCalled = true
+            return ch.readAvailable(sink, offset, length)
+        }
+    }
+
+    @Test
+    fun testReadAvailableSdkByteBufferFallback() = runTest {
+        val content = "a".repeat(64).encodeToByteArray()
+        val inner = SdkByteReadChannel(content)
+        val chan = ProxyChan(inner)
+
+        val dest = SdkByteBuffer(32U)
+        val rc = chan.readAvailable(dest)
+        assertEquals(dest.readRemaining.toLong(), rc)
+        val expected = "a".repeat(32)
+        assertEquals(expected, dest.decodeToString())
+        assertTrue(chan.proxyCalled)
+    }
 }

runtime/protocol/http-client-engines/test-suite/common/test/aws/smithy/kotlin/runtime/http/test/UploadTest.kt

@@ -5,6 +5,7 @@
 
 package aws.smithy.kotlin.runtime.http.test
 
+import aws.smithy.kotlin.runtime.content.ByteStream
 import aws.smithy.kotlin.runtime.http.HttpBody
 import aws.smithy.kotlin.runtime.http.HttpMethod
 import aws.smithy.kotlin.runtime.http.HttpStatusCode
@@ -14,6 +15,7 @@ import aws.smithy.kotlin.runtime.http.request.url
 import aws.smithy.kotlin.runtime.http.response.complete
 import aws.smithy.kotlin.runtime.http.test.util.AbstractEngineTest
 import aws.smithy.kotlin.runtime.http.test.util.test
+import aws.smithy.kotlin.runtime.http.toHttpBody
 import aws.smithy.kotlin.runtime.io.SdkByteChannel
 import aws.smithy.kotlin.runtime.io.SdkByteReadChannel
 import aws.smithy.kotlin.runtime.util.encodeToHex
@@ -29,7 +31,7 @@ class UploadTest : AbstractEngineTest() {
     fun testUploadIntegrity() = testEngines {
         // test that what we write the entire contents given to us
         test { env, client ->
-            val data = ByteArray(16 * 1024 * 1023) { it.toByte() }
+            val data = ByteArray(16 * 1024 * 1024) { it.toByte() }
             val sha = data.sha256().encodeToHex()
 
             val req = HttpRequest {
@@ -80,4 +82,34 @@ class UploadTest : AbstractEngineTest() {
             }
         }
     }
+
+    @Test
+    fun testUploadWithWrappedStream() = testEngines {
+        // test custom ByteStream behavior
+        // see https://github.com/awslabs/smithy-kotlin/issues/613
+        test { env, client ->
+            val data = ByteArray(1024 * 1024) { it.toByte() }
+            val sha = data.sha256().encodeToHex()
+
+            val wrappedStream = object : ByteStream.ReplayableStream() {
+                override val contentLength: Long = data.size.toLong()
+                override fun newReader(): SdkByteReadChannel {
+                    val underlying = SdkByteReadChannel(data)
+                    return object : SdkByteReadChannel by underlying {}
+                }
+            }
+
+            val req = HttpRequest {
+                method = HttpMethod.POST
+                url(env.testServer)
+                url.path = "/upload/content"
+                body = wrappedStream.toHttpBody()
+            }
+
+            val call = client.call(req)
+            call.complete()
+            assertEquals(HttpStatusCode.OK, call.response.status)
+            assertEquals(sha, call.response.headers["content-sha256"], "sha mismatch for upload on ${client.engine}")
+        }
+    }
 }