fix: handle stream response exceptions properly in Ktor engine

runtime/protocol/http-client-engines/http-client-engine-ktor/build.gradle.kts

@@ -7,6 +7,7 @@ description = "Ktor HTTP Client Engine for Smithy services generated by smithy-k
 extra["displayName"] = "Smithy :: Kotlin :: HTTP :: Engine :: Ktor"
 extra["moduleName"] = "aws.smithy.kotlin.runtime.http.engine.ktor"
 
+val coroutinesVersion: String by project
 val ktorVersion: String by project
 
 kotlin {
@@ -24,6 +25,11 @@ kotlin {
                 implementation(project(":runtime:logging"))
             }
         }
+        jvmTest {
+            dependencies {
+                implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:$coroutinesVersion")
+            }
+        }
 
         all {
             languageSettings.optIn("aws.smithy.kotlin.runtime.util.InternalApi")

runtime/protocol/http-client-engines/http-client-engine-ktor/jvm/src/aws/smithy/kotlin/runtime/http/engine/ktor/KtorEngine.kt

@@ -21,6 +21,7 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
 import okhttp3.ConnectionPool
 import okhttp3.Protocol
 import java.util.concurrent.TimeUnit
@@ -152,11 +153,11 @@ actual class KtorEngine actual constructor(
  * Simple notify mechanism that waits for a signal
  */
 internal class Waiter {
-    private val channel = Channel<Unit>(0)
+    private val mutex = Mutex(locked = true)
 
     // wait for the signal
-    suspend fun wait() { channel.receive() }
+    suspend fun wait() { mutex.lock() }
 
     // give the signal to continue
-    fun signal() { channel.trySend(Unit).getOrThrow() }
+    fun signal() { mutex.unlock() }
 }

runtime/protocol/http-client-engines/http-client-engine-ktor/jvm/test/aws/smithy/kotlin/runtime/http/engine/ktor/KtorEngineTest.kt

@@ -0,0 +1,46 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+package aws.smithy.kotlin.runtime.http.engine.ktor
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runBlockingTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.ExperimentalTime
+
+@OptIn(ExperimentalCoroutinesApi::class, ExperimentalTime::class)
+class WaiterTest {
+    @Test
+    fun testSignalWhenWaiting() = runBlockingTest {
+        val start = currentTime
+
+        val waiter = Waiter()
+        launch {
+            delay(500.milliseconds)
+            waiter.signal()
+        }
+        waiter.wait()
+
+        assertEquals(500, currentTime - start)
+    }
+
+    @Test
+    fun testSignalWhenNotWaiting() = runBlockingTest {
+        val start = currentTime
+
+        val waiter = Waiter()
+        launch {
+            delay(500.milliseconds)
+            waiter.signal()
+        }
+        delay(1000.milliseconds)
+
+        assertEquals(1000, currentTime - start)
+    }
+}