fix: Close CRT HttpStream on call completion (#1134)

Author: lauzadis

.changes/1ab4f8b7-5721-4024-a4dc-d0c595495bd2.json

@@ -0,0 +1,5 @@
+{
+    "id": "1ab4f8b7-5721-4024-a4dc-d0c595495bd2",
+    "type": "bugfix",
+    "description": "Close HttpStreams in CrtHttpEngine after call completion"
+}

runtime/protocol/http-client-engines/http-client-engine-crt/jvm/src/aws/smithy/kotlin/runtime/http/engine/crt/CrtHttpEngine.kt

@@ -79,6 +79,10 @@ public class CrtHttpEngine(public override val config: CrtHttpEngineConfig) : Ht
             }
         }
 
+        callContext.job.invokeOnCompletion {
+            stream.close()
+        }
+
         if (request.isChunked) {
             withContext(SdkDispatchers.IO) {
                 stream.sendChunkedBody(request.body)

runtime/protocol/http-client-engines/http-client-engine-crt/jvm/src/aws/smithy/kotlin/runtime/http/engine/crt/SdkStreamResponseHandler.kt

@@ -149,7 +149,11 @@ internal class SdkStreamResponseHandler(
         }
 
         // short circuit, stop buffering data and discard remaining incoming bytes
-        if (isCancelled) return bodyBytesIn.len
+        if (isCancelled) {
+            crtStream?.close()
+            stream.close()
+            return bodyBytesIn.len
+        }
 
         val buffer = SdkBuffer().apply {
             val bytes = bodyBytesIn.readAll()
@@ -166,9 +170,11 @@ internal class SdkStreamResponseHandler(
         // stream is only valid until the end of this callback, ensure any further data being read downstream
         // doesn't call incrementWindow on a resource that has been free'd
         lock.withLock {
+            crtStream?.close()
             crtStream = null
             streamCompleted = true
         }
+        stream.close()
 
         // close the body channel
         if (errorCode != 0) {