diff --git a/http-client/http-client-java/src/main/kotlin/ai/koog/http/client/java/JavaKoogHttpClient.kt b/http-client/http-client-java/src/main/kotlin/ai/koog/http/client/java/JavaKoogHttpClient.kt index ff5c5fa694..990287e6f4 100644 --- a/http-client/http-client-java/src/main/kotlin/ai/koog/http/client/java/JavaKoogHttpClient.kt +++ b/http-client/http-client-java/src/main/kotlin/ai/koog/http/client/java/JavaKoogHttpClient.kt @@ -31,6 +31,8 @@ import kotlin.time.Duration.Companion.minutes * This client provides enhanced logging, flexible request and response handling, and supports * configurability for underlying Java HttpClient instances. * + * SSE dispatches complete UTF-8 data events, preserving multiline payloads and ignoring event metadata. + * * @property clientName The name of the client, used for logging and traceability. * @property logger A logging instance of type KLogger for recording client-related events and errors. * @property httpClient The configured Java HttpClient instance used for making HTTP requests. @@ -203,24 +205,27 @@ public class JavaKoogHttpClient internal constructor( logger.debug { "SSE connection opened for $clientName" } - // Process the stream of lines - response.body().forEach { line -> + val data = StringBuilder() + var firstLine = true + response.body().forEach { rawLine -> try { - val dataPrefix = "data: " - // SSE format: "data: " - val data = if (line.startsWith(dataPrefix)) { - line.substring(dataPrefix.length) - } else if (line.isNotEmpty() && !line.startsWith(":")) { - line - } else { - null - } - - if (data != null && dataFilter(data)) { - data.trim() - .let(decodeStreamingResponse) - .let(processStreamingChunk) - ?.let { trySend(it) } + val line = if (firstLine) rawLine.removePrefix("\uFEFF") else rawLine + firstLine = false + if (line.isEmpty()) { + if (data.isNotEmpty()) { + val payload = data.dropLast(1).toString() + data.setLength(0) + if (dataFilter(payload)) { + processStreamingChunk(decodeStreamingResponse(payload))?.let { trySend(it) } + } + } + } else if (!line.startsWith(":")) { + val colon = line.indexOf(':') + val field = if (colon < 0) line else line.substring(0, colon) + if (field == "data") { + val value = if (colon < 0) "" else line.substring(colon + 1).removePrefix(" ") + data.append(value).append('\n') + } } } catch (e: CancellationException) { throw e @@ -235,6 +240,7 @@ public class JavaKoogHttpClient internal constructor( } } + // Pending data without a terminating empty line belongs to an incomplete event. logger.debug { "SSE connection closed for $clientName" } close() } catch (e: CancellationException) { diff --git a/http-client/http-client-java/src/test/kotlin/ai/koog/http/client/java/JavaKoogHttpClientSseFramingTest.kt b/http-client/http-client-java/src/test/kotlin/ai/koog/http/client/java/JavaKoogHttpClientSseFramingTest.kt new file mode 100644 index 0000000000..8c10640f0a --- /dev/null +++ b/http-client/http-client-java/src/test/kotlin/ai/koog/http/client/java/JavaKoogHttpClientSseFramingTest.kt @@ -0,0 +1,97 @@ +package ai.koog.http.client.java + +import ai.koog.http.client.KoogHttpClient +import ai.koog.http.client.test.MockWebServer +import io.ktor.http.ContentType +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.flow.flowOn +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals + +class JavaKoogHttpClientSseFramingTest { + @Test + fun testMultilineDataDispatchesAsOneEvent() = runTest { + assertEquals(listOf("first\nsecond"), collect("data: first\ndata:second\n\n")) + } + + @Test + fun testMetadataAndCommentsAreIgnored() = runTest { + assertEquals(listOf("payload"), collect(": heartbeat\nevent: message\nid: 7\nretry: 100\nfuture: ignored\nraw text\ndata: payload\n\n")) + } + + @Test + fun testLeadingBomAndAllLineEndings() = runTest { + assertEquals(listOf("one", "two", "three"), collect("\uFEFFdata: one\r\rdata: two\r\n\r\ndata: three\n\n")) + } + + @Test + fun testDataWhitespaceIsPreserved() = runTest { + assertEquals(listOf(" leading \n\ttab "), collect("data: leading \ndata:\ttab \n\n")) + } + + @Test + fun testEmptyDataAndColonlessDataDispatch() = runTest { + assertEquals(listOf("", ""), collect("\nevent: ignored\n\ndata:\n\ndata\n\n")) + } + + @Test + fun testIncompleteEventIsDiscardedAtEof() = runTest { + assertEquals(listOf("complete"), collect("data: complete\n\ndata: incomplete\n")) + } + + @Test + fun testFilterAndDecoderReceiveCompleteDataEvents() = runTest { + val filtered = mutableListOf() + withResponse("data: first\ndata: second\n\ndata: skip\n\ndata: keep\n\n") { client -> + val result = client.sse( + path = "/stream", + requestBody = "{}", + requestBodyType = String::class, + dataFilter = { + filtered.add(it) + it != "skip" + }, + decodeStreamingResponse = { it.uppercase() }, + processStreamingChunk = { it.takeUnless { it == "FIRST\nSECOND" } }, + ).flowOn(Dispatchers.IO).toList() + assertEquals(listOf("first\nsecond", "skip", "keep"), filtered) + assertEquals(listOf("KEEP"), result) + } + } + + private suspend fun collect(response: String): List { + var result = emptyList() + withResponse(response) { client -> + result = client.sse( + path = "/stream", + requestBody = "{}", + requestBodyType = String::class, + decodeStreamingResponse = { it }, + processStreamingChunk = { it }, + ).flowOn(Dispatchers.IO).toList() + } + return result + } + + private suspend fun withResponse(response: String, block: suspend (KoogHttpClient) -> Unit) { + val server = MockWebServer() + server.start( + postEndpoints = listOf( + MockWebServer.PostEndpointConfig( + path = "/stream", + responseBody = response, + contentType = ContentType.Text.EventStream, + ) + ) + ) + val client = JavaKoogHttpClient.Factory().create(clientName = "java-sse", baseUrl = server.url("")) + try { + block(client) + } finally { + client.close() + server.stop() + } + } +}