KTOR-6482 Logging plugin blocks response body streaming when level is…

… BODY (#3843)
ktorio · Nov 21, 2023 · e468c64 · e468c64
1 parent 9a67d61
commit e468c64
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 6 deletions.
diff --git a/ktor-client/ktor-client-tests/common/test/io/ktor/client/tests/LoggingMockedTests.kt b/ktor-client/ktor-client-tests/common/test/io/ktor/client/tests/LoggingMockedTests.kt
@@ -13,7 +13,10 @@ import io.ktor.client.statement.*
 import io.ktor.client.tests.utils.*
 import io.ktor.http.*
 import io.ktor.util.*
+import io.ktor.utils.io.*
 import io.ktor.utils.io.core.*
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
 import kotlin.test.*
 
 class LoggingMockedTests {
@@ -339,4 +342,44 @@ class LoggingMockedTests {
             testLogger.verify()
         }
     }
+
+    @Test
+    fun testCanStream() = testWithEngine(MockEngine) {
+        val channel = ByteChannel(autoFlush = true)
+        config {
+            engine {
+                addHandler {
+                    respond(
+                        content = channel,
+                        status = HttpStatusCode.OK
+                    )
+                }
+            }
+            install(Logging) {
+                level = LogLevel.BODY
+                logger = Logger.DEFAULT
+            }
+        }
+        test { client ->
+            val content = channelFlow {
+                launch {
+                    client.preparePost("/").execute {
+                        val ch = it.bodyAsChannel()
+                        while (!ch.isClosedForRead) {
+                            ch.awaitContent()
+                            send(ch.readUTF8Line())
+                        }
+                    }
+                }
+            }
+
+            channel.writeStringUtf8("Hello world!\n")
+
+            withTimeout(5_000) { // the bug will cause this to timeout
+                content.collect {
+                    channel.close()
+                }
+            }
+        }
+    }
 }
diff --git a/ktor-utils/common/src/io/ktor/util/ByteChannels.kt b/ktor-utils/common/src/io/ktor/util/ByteChannels.kt
@@ -6,6 +6,7 @@ package io.ktor.util
 
 import io.ktor.utils.io.*
 import io.ktor.utils.io.core.*
+import io.ktor.utils.io.pool.*
 import kotlinx.coroutines.*
 
 private const val CHUNK_BUFFER_SIZE = 4096L
@@ -20,14 +21,14 @@ public fun ByteReadChannel.split(coroutineScope: CoroutineScope): Pair<ByteReadC
     val second = ByteChannel(autoFlush = true)
 
     coroutineScope.launch {
+        val buffer = ByteArrayPool.borrow()
         try {
             while (!isClosedForRead) {
-                this@split.readRemaining(CHUNK_BUFFER_SIZE).use { chunk ->
-                    listOf(
-                        async { first.writePacket(chunk.copy()) },
-                        async { second.writePacket(chunk.copy()) }
-                    ).awaitAll()
-                }
+                val read = this@split.readAvailable(buffer)
+                listOf(
+                    async { first.writeFully(buffer, 0, read) },
+                    async { second.writeFully(buffer, 0, read) }
+                ).awaitAll()
             }
 
             closedCause?.let { throw it }
@@ -36,6 +37,7 @@ public fun ByteReadChannel.split(coroutineScope: CoroutineScope): Pair<ByteReadC
             first.cancel(cause)
             second.cancel(cause)
         } finally {
+            ByteArrayPool.recycle(buffer)
             first.close()
             second.close()
         }