KTOR-8204 Fix for exception on empty chunked body response

bjhham · bjhham · commit 56215409ed09 · 2025-04-28T17:06:49.000+02:00
diff --git a/ktor-client/ktor-client-core/common/src/io/ktor/client/plugins/sse/DefaultClientSSESession.kt b/ktor-client/ktor-client-core/common/src/io/ktor/client/plugins/sse/DefaultClientSSESession.kt
@@ -45,7 +45,7 @@ public class DefaultClientSSESession(
         // we have an outer while to obtain new input
         while (this@DefaultClientSSESession.coroutineContext.isActive) {
             while (this@DefaultClientSSESession.coroutineContext.isActive) {
-                val event = input.parseEvent() ?: break
+                val event = input.tryParseEvent() ?: break
 
                 if (event.isCommentsEvent() && !showCommentEvents) continue
                 if (event.isRetryEvent() && !showRetryEvents) continue
@@ -140,6 +140,14 @@ public class DefaultClientSSESession(
         input.cancel()
     }
 
+    private suspend fun ByteReadChannel.tryParseEvent(): ServerSentEvent? =
+        try {
+            parseEvent()
+        } catch (_: ClosedByteChannelException) {
+            // this is expected when the server disconnects
+            null
+        }
+
     private suspend fun ByteReadChannel.parseEvent(): ServerSentEvent? {
         val data = StringBuilder()
         val comments = StringBuilder()
diff --git a/ktor-client/ktor-client-tests/common/test/io/ktor/client/tests/plugins/ServerSentEventsTest.kt b/ktor-client/ktor-client-tests/common/test/io/ktor/client/tests/plugins/ServerSentEventsTest.kt
@@ -19,7 +19,6 @@ import io.ktor.sse.*
 import io.ktor.utils.io.*
 import io.ktor.utils.io.charsets.*
 import kotlinx.coroutines.*
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.collectIndexed
 import kotlinx.coroutines.flow.single
 import kotlinx.coroutines.flow.take
@@ -33,8 +32,6 @@ import kotlin.coroutines.resume
 import kotlin.coroutines.suspendCoroutine
 import kotlin.test.*
 import kotlin.time.Duration.Companion.milliseconds
-import kotlin.time.Duration.Companion.seconds
-import kotlin.time.measureTime
 
 class ServerSentEventsTest : ClientLoader() {
 
@@ -138,19 +135,23 @@ class ServerSentEventsTest : ClientLoader() {
 
         test { client ->
             coroutineScope {
-                val job = launch {
-                    val session = client.serverSentEventsSession("$TEST_SERVER/sse/hello?times=20&interval=100")
+                val started = Job()
+                val session = client.serverSentEventsSession("$TEST_SERVER/sse/hello?times=20&interval=100")
+                val readJob = launch {
                     try {
-                        session.incoming.collect()
+                        session.incoming.collect {
+                            started.complete()
+                        }
                     } finally {
                         withContext(NonCancellable) {
-                            delay(500)
+                            started.join()
                             assertFalse(session.isActive)
                         }
                     }
                 }
-                delay(500)
-                job.cancelAndJoin()
+                started.join()
+                readJob.cancelAndJoin()
+                assertFalse(session.isActive)
             }
         }
     }
@@ -582,25 +583,22 @@ class ServerSentEventsTest : ClientLoader() {
         config {
             install(SSE) {
                 maxReconnectionAttempts = 1
-                reconnectionTime = 2.seconds
+                reconnectionTime = 100.milliseconds
             }
         }
 
         test { client ->
             val events = mutableListOf<ServerSentEvent>()
 
-            val time = measureTime {
-                client.sse("$TEST_SERVER/sse/reconnection?count=5") {
-                    incoming.take(10).collect {
-                        events.add(it)
-                    }
+            client.sse("$TEST_SERVER/sse/reconnection?count=5") {
+                incoming.take(10).collect {
+                    events.add(it)
                 }
             }
 
             events.forEachIndexed { index, event ->
                 assertEquals(index + 1, event.id?.toInt())
             }
-            assertTrue { time > 2.seconds }
         }
     }
 
@@ -617,7 +615,7 @@ class ServerSentEventsTest : ClientLoader() {
             var count = 0
 
             assertFailsWith<IllegalStateException> {
-                client.sse("$TEST_SERVER/sse/reconnection?count=5", reconnectionTime = 1.seconds) {
+                client.sse("$TEST_SERVER/sse/reconnection?count=5", reconnectionTime = 10.milliseconds) {
                     incoming.collect {
                         events.add(it)
                         count++
@@ -669,7 +667,6 @@ class ServerSentEventsTest : ClientLoader() {
     fun testSeveralReconnections() = clientTests(except("OkHttp")) {
         config {
             install(SSE) {
-                reconnectionTime = 500.milliseconds
                 maxReconnectionAttempts = 2
             }
         }
@@ -678,14 +675,12 @@ class ServerSentEventsTest : ClientLoader() {
             val events = mutableListOf<ServerSentEvent>()
             var count = 0
 
-            val time = measureTime {
-                client.sse("$TEST_SERVER/sse/reconnection?count=5") {
-                    incoming.collect {
-                        events.add(it)
-                        count++
-                        if (count == 15) {
-                            cancel()
-                        }
+            client.sse("$TEST_SERVER/sse/reconnection?count=5", reconnectionTime = 10.milliseconds) {
+                incoming.collect {
+                    events.add(it)
+                    count++
+                    if (count == 15) {
+                        cancel()
                     }
                 }
             }
@@ -694,15 +689,14 @@ class ServerSentEventsTest : ClientLoader() {
             events.forEachIndexed { index, event ->
                 assertEquals(index + 1, event.id?.toInt())
             }
-            assertTrue { 1.seconds < time && time < 2.seconds }
         }
     }
 
     @Test
     fun testMaxRetries() = clientTests(except("OkHttp")) {
         config {
             install(SSE) {
-                reconnectionTime = 500.milliseconds
+                reconnectionTime = 10.milliseconds
                 maxReconnectionAttempts = 4
             }
         }
@@ -711,14 +705,12 @@ class ServerSentEventsTest : ClientLoader() {
             val events = mutableListOf<ServerSentEvent>()
             var count = 0
 
-            val time = measureTime {
-                client.sse("$TEST_SERVER/sse/exception-on-reconnection?count=5&count-of-reconnections=4") {
-                    incoming.collect {
-                        events.add(it)
-                        count++
-                        if (count == 10) {
-                            cancel()
-                        }
+            client.sse("$TEST_SERVER/sse/exception-on-reconnection?count=5&count-of-reconnections=4") {
+                incoming.collect {
+                    events.add(it)
+                    count++
+                    if (count == 10) {
+                        cancel()
                     }
                 }
             }
@@ -727,7 +719,6 @@ class ServerSentEventsTest : ClientLoader() {
             events.forEachIndexed { index, event ->
                 assertEquals(index % 5, event.id?.toInt())
             }
-            assertTrue { 2.seconds < time && time < 3.seconds }
         }
     }
 
@@ -741,7 +732,6 @@ class ServerSentEventsTest : ClientLoader() {
         }
 
         test { client ->
-
             client.sse("$TEST_SERVER/sse/no-content") {
                 assertEquals(HttpStatusCode.NoContent, call.response.status)
                 assertEquals(0, incoming.toList().size)
@@ -759,4 +749,18 @@ class ServerSentEventsTest : ClientLoader() {
             }
         }
     }
+
+    @Test
+    fun testNoContentStream() = clientTests(except("OkHttp")) {
+        config {
+            install(SSE)
+        }
+
+        test { client ->
+            client.sse("$TEST_SERVER/sse/no-events") {
+                assertEquals(HttpStatusCode.OK, call.response.status)
+                assertEquals(0, incoming.toList().size)
+            }
+        }
+    }
 }
diff --git a/ktor-http/ktor-http-cio/common/src/io/ktor/http/cio/ChunkedTransferEncoding.kt b/ktor-http/ktor-http-cio/common/src/io/ktor/http/cio/ChunkedTransferEncoding.kt
@@ -69,11 +69,8 @@ public suspend fun decodeChunked(input: ByteReadChannel, out: ByteWriteChannel)
     var totalBytesCopied = 0L
 
     try {
-        while (true) {
-            chunkSizeBuffer.clear()
-            if (!input.readUTF8LineTo(chunkSizeBuffer, MAX_CHUNK_SIZE_LENGTH, httpLineEndings)) {
-                throw EOFException("Chunked stream has ended unexpectedly: no chunk size")
-            } else if (chunkSizeBuffer.isEmpty()) {
+        while (input.readUTF8LineTo(chunkSizeBuffer, MAX_CHUNK_SIZE_LENGTH, httpLineEndings)) {
+            if (chunkSizeBuffer.isEmpty()) {
                 throw EOFException("Invalid chunk size: empty")
             }
 
diff --git a/ktor-http/ktor-http-cio/jvm/test/io/ktor/tests/http/cio/ChunkedTest.kt b/ktor-http/ktor-http-cio/jvm/test/io/ktor/tests/http/cio/ChunkedTest.kt
@@ -24,17 +24,6 @@ import kotlin.test.assertTrue
 
 class ChunkedTest {
 
-    @Test
-    fun testEmptyBroken(): Unit = runBlocking {
-        val bodyText = ""
-        val ch = ByteReadChannel(bodyText.toByteArray())
-        val parsed = ByteChannel()
-
-        assertFailsWith<EOFException> {
-            decodeChunked(ch, parsed)
-        }
-    }
-
     @Test
     fun testChunkedWithContentLength() = runBlocking {
         val chunkedContent = listOf(
@@ -71,6 +60,16 @@ class ChunkedTest {
         }
     }
 
+    @Test
+    fun testEmptyNoCRLF(): Unit = runBlocking {
+        val bodyText = ""
+        val ch = ByteReadChannel(bodyText.toByteArray())
+        val parsed = ByteChannel()
+
+        decodeChunked(ch, parsed)
+        assertEquals("", parsed.readBuffer().readText())
+    }
+
     @Test
     fun testEmpty() = runBlocking {
         val bodyText = "0\r\n\r\n"
diff --git a/ktor-test-server/src/main/kotlin/test/server/tests/ServerSentEvents.kt b/ktor-test-server/src/main/kotlin/test/server/tests/ServerSentEvents.kt
@@ -153,6 +153,9 @@ internal fun Application.serverSentEvents() {
             get("no-content") {
                 call.respond(HttpStatusCode.NoContent)
             }
+            get("no-events") {
+                call.respondBytesWriter(ContentType.Text.EventStream) {}
+            }
             get("no-content-after-reconnection") {
                 val count = call.parameters["count"]?.toInt() ?: 0
                 val lastEventId = call.request.header("Last-Event-ID")

Original file line number	Diff line number	Diff line change
`@@ -19,7 +19,6 @@ import io.ktor.sse.*`
`19`	`19`	`import io.ktor.utils.io.*`
`20`	`20`	`import io.ktor.utils.io.charsets.*`
`21`	`21`	`import kotlinx.coroutines.*`
`22`		`-import kotlinx.coroutines.flow.collect`
`23`	`22`	`import kotlinx.coroutines.flow.collectIndexed`
`24`	`23`	`import kotlinx.coroutines.flow.single`
`25`	`24`	`import kotlinx.coroutines.flow.take`
`@@ -33,8 +32,6 @@ import kotlin.coroutines.resume`
`33`	`32`	`import kotlin.coroutines.suspendCoroutine`
`34`	`33`	`import kotlin.test.*`
`35`	`34`	`import kotlin.time.Duration.Companion.milliseconds`
`36`		`-import kotlin.time.Duration.Companion.seconds`
`37`		`-import kotlin.time.measureTime`
`38`	`35`
`39`	`36`	`class ServerSentEventsTest : ClientLoader() {`
`40`	`37`
`@@ -138,19 +135,23 @@ class ServerSentEventsTest : ClientLoader() {`
`138`	`135`
`139`	`136`	`test { client ->`
`140`	`137`	`coroutineScope {`
`141`		`- val job = launch {`
`142`		`- val session = client.serverSentEventsSession("$TEST_SERVER/sse/hello?times=20&interval=100")`
	`138`	`+ val started = Job()`
	`139`	`+ val session = client.serverSentEventsSession("$TEST_SERVER/sse/hello?times=20&interval=100")`
	`140`	`+ val readJob = launch {`
`143`	`141`	`try {`
`144`		`- session.incoming.collect()`
	`142`	`+ session.incoming.collect {`
	`143`	`+ started.complete()`
	`144`	`+ }`
`145`	`145`	`} finally {`
`146`	`146`	`withContext(NonCancellable) {`
`147`		`- delay(500)`
	`147`	`+ started.join()`
`148`	`148`	`assertFalse(session.isActive)`
`149`	`149`	`}`
`150`	`150`	`}`
`151`	`151`	`}`
`152`		`- delay(500)`
`153`		`- job.cancelAndJoin()`
	`152`	`+ started.join()`
	`153`	`+ readJob.cancelAndJoin()`
	`154`	`+ assertFalse(session.isActive)`
`154`	`155`	`}`
`155`	`156`	`}`
`156`	`157`	`}`
`@@ -582,25 +583,22 @@ class ServerSentEventsTest : ClientLoader() {`
`582`	`583`	`config {`
`583`	`584`	`install(SSE) {`
`584`	`585`	`maxReconnectionAttempts = 1`
`585`		`- reconnectionTime = 2.seconds`
	`586`	`+ reconnectionTime = 100.milliseconds`
`586`	`587`	`}`
`587`	`588`	`}`
`588`	`589`
`589`	`590`	`test { client ->`
`590`	`591`	`val events = mutableListOf<ServerSentEvent>()`
`591`	`592`
`592`		`- val time = measureTime {`
`593`		`- client.sse("$TEST_SERVER/sse/reconnection?count=5") {`
`594`		`- incoming.take(10).collect {`
`595`		`- events.add(it)`
`596`		`- }`
	`593`	`+ client.sse("$TEST_SERVER/sse/reconnection?count=5") {`
	`594`	`+ incoming.take(10).collect {`
	`595`	`+ events.add(it)`
`597`	`596`	`}`
`598`	`597`	`}`
`599`	`598`
`600`	`599`	`events.forEachIndexed { index, event ->`
`601`	`600`	`assertEquals(index + 1, event.id?.toInt())`
`602`	`601`	`}`
`603`		`- assertTrue { time > 2.seconds }`
`604`	`602`	`}`
`605`	`603`	`}`
`606`	`604`
`@@ -617,7 +615,7 @@ class ServerSentEventsTest : ClientLoader() {`
`617`	`615`	`var count = 0`
`618`	`616`
`619`	`617`	`assertFailsWith<IllegalStateException> {`
`620`		`- client.sse("$TEST_SERVER/sse/reconnection?count=5", reconnectionTime = 1.seconds) {`
	`618`	`+ client.sse("$TEST_SERVER/sse/reconnection?count=5", reconnectionTime = 10.milliseconds) {`
`621`	`619`	`incoming.collect {`
`622`	`620`	`events.add(it)`
`623`	`621`	`count++`
`@@ -669,7 +667,6 @@ class ServerSentEventsTest : ClientLoader() {`
`669`	`667`	`fun testSeveralReconnections() = clientTests(except("OkHttp")) {`
`670`	`668`	`config {`
`671`	`669`	`install(SSE) {`
`672`		`- reconnectionTime = 500.milliseconds`
`673`	`670`	`maxReconnectionAttempts = 2`
`674`	`671`	`}`
`675`	`672`	`}`
`@@ -678,14 +675,12 @@ class ServerSentEventsTest : ClientLoader() {`
`678`	`675`	`val events = mutableListOf<ServerSentEvent>()`
`679`	`676`	`var count = 0`
`680`	`677`
`681`		`- val time = measureTime {`
`682`		`- client.sse("$TEST_SERVER/sse/reconnection?count=5") {`
`683`		`- incoming.collect {`
`684`		`- events.add(it)`
`685`		`- count++`
`686`		`- if (count == 15) {`
`687`		`- cancel()`
`688`		`- }`
	`678`	`+ client.sse("$TEST_SERVER/sse/reconnection?count=5", reconnectionTime = 10.milliseconds) {`
	`679`	`+ incoming.collect {`
	`680`	`+ events.add(it)`
	`681`	`+ count++`
	`682`	`+ if (count == 15) {`
	`683`	`+ cancel()`
`689`	`684`	`}`
`690`	`685`	`}`
`691`	`686`	`}`
`@@ -694,15 +689,14 @@ class ServerSentEventsTest : ClientLoader() {`
`694`	`689`	`events.forEachIndexed { index, event ->`
`695`	`690`	`assertEquals(index + 1, event.id?.toInt())`
`696`	`691`	`}`
`697`		`- assertTrue { 1.seconds < time && time < 2.seconds }`
`698`	`692`	`}`
`699`	`693`	`}`
`700`	`694`
`701`	`695`	`@Test`
`702`	`696`	`fun testMaxRetries() = clientTests(except("OkHttp")) {`
`703`	`697`	`config {`
`704`	`698`	`install(SSE) {`
`705`		`- reconnectionTime = 500.milliseconds`
	`699`	`+ reconnectionTime = 10.milliseconds`
`706`	`700`	`maxReconnectionAttempts = 4`
`707`	`701`	`}`
`708`	`702`	`}`
`@@ -711,14 +705,12 @@ class ServerSentEventsTest : ClientLoader() {`
`711`	`705`	`val events = mutableListOf<ServerSentEvent>()`
`712`	`706`	`var count = 0`
`713`	`707`
`714`		`- val time = measureTime {`
`715`		`- client.sse("$TEST_SERVER/sse/exception-on-reconnection?count=5&count-of-reconnections=4") {`
`716`		`- incoming.collect {`
`717`		`- events.add(it)`
`718`		`- count++`
`719`		`- if (count == 10) {`
`720`		`- cancel()`
`721`		`- }`
	`708`	`+ client.sse("$TEST_SERVER/sse/exception-on-reconnection?count=5&count-of-reconnections=4") {`
	`709`	`+ incoming.collect {`
	`710`	`+ events.add(it)`
	`711`	`+ count++`
	`712`	`+ if (count == 10) {`
	`713`	`+ cancel()`
`722`	`714`	`}`
`723`	`715`	`}`
`724`	`716`	`}`
`@@ -727,7 +719,6 @@ class ServerSentEventsTest : ClientLoader() {`
`727`	`719`	`events.forEachIndexed { index, event ->`
`728`	`720`	`assertEquals(index % 5, event.id?.toInt())`
`729`	`721`	`}`
`730`		`- assertTrue { 2.seconds < time && time < 3.seconds }`
`731`	`722`	`}`
`732`	`723`	`}`
`733`	`724`
`@@ -741,7 +732,6 @@ class ServerSentEventsTest : ClientLoader() {`
`741`	`732`	`}`
`742`	`733`
`743`	`734`	`test { client ->`
`744`		`-`
`745`	`735`	`client.sse("$TEST_SERVER/sse/no-content") {`
`746`	`736`	`assertEquals(HttpStatusCode.NoContent, call.response.status)`
`747`	`737`	`assertEquals(0, incoming.toList().size)`
`@@ -759,4 +749,18 @@ class ServerSentEventsTest : ClientLoader() {`
`759`	`749`	`}`
`760`	`750`	`}`
`761`	`751`	`}`
	`752`	`+`
	`753`	`+ @Test`
	`754`	`+ fun testNoContentStream() = clientTests(except("OkHttp")) {`
	`755`	`+ config {`
	`756`	`+ install(SSE)`
	`757`	`+ }`
	`758`	`+`
	`759`	`+ test { client ->`
	`760`	`+ client.sse("$TEST_SERVER/sse/no-events") {`
	`761`	`+ assertEquals(HttpStatusCode.OK, call.response.status)`
	`762`	`+ assertEquals(0, incoming.toList().size)`
	`763`	`+ }`
	`764`	`+ }`
	`765`	`+ }`
`762`	`766`	`}`