Upgrade Ktor version to stable v3

app/src/main/kotlin/dev/chungjungsoo/gptmobile/data/network/AnthropicAPIImpl.kt

@@ -5,26 +5,15 @@ import dev.chungjungsoo.gptmobile.data.dto.anthropic.request.MessageRequest
 import dev.chungjungsoo.gptmobile.data.dto.anthropic.response.ErrorDetail
 import dev.chungjungsoo.gptmobile.data.dto.anthropic.response.ErrorResponseChunk
 import dev.chungjungsoo.gptmobile.data.dto.anthropic.response.MessageResponseChunk
-import io.ktor.client.call.body
-import io.ktor.client.request.HttpRequestBuilder
+import io.ktor.client.plugins.sse.sse
 import io.ktor.client.request.accept
 import io.ktor.client.request.headers
 import io.ktor.client.request.setBody
-import io.ktor.client.request.url
-import io.ktor.client.statement.HttpResponse
-import io.ktor.client.statement.HttpStatement
 import io.ktor.http.ContentType
 import io.ktor.http.HttpMethod
-import io.ktor.http.contentType
-import io.ktor.utils.io.ByteReadChannel
-import io.ktor.utils.io.cancel
-import io.ktor.utils.io.readUTF8Line
 import javax.inject.Inject
-import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.FlowCollector
 import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.isActive
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.encodeToJsonElement
 
@@ -43,52 +32,29 @@ class AnthropicAPIImpl @Inject constructor(
         this.apiUrl = url
     }
 
-    override fun streamChatMessage(messageRequest: MessageRequest): Flow<MessageResponseChunk> {
-        val body = Json.encodeToJsonElement(messageRequest)
-
-        val builder = HttpRequestBuilder().apply {
-            method = HttpMethod.Post
-            if (apiUrl.endsWith("/")) url("${apiUrl}v1/messages") else url("$apiUrl/v1/messages")
-            contentType(ContentType.Application.Json)
-            setBody(body)
-            accept(ContentType.Text.EventStream)
-            headers {
-                append(API_KEY_HEADER, token ?: "")
-                append(VERSION_HEADER, ANTHROPIC_VERSION)
-            }
-        }
-
-        return flow {
-            try {
-                HttpStatement(builder = builder, client = networkClient()).execute {
-                    streamEventsFrom(it)
-                }
-            } catch (e: Exception) {
-                emit(ErrorResponseChunk(error = ErrorDetail(type = "network_error", message = e.message ?: "")))
-            }
-        }
-    }
-
-    private suspend inline fun <reified T> FlowCollector<T>.streamEventsFrom(response: HttpResponse) {
-        val channel: ByteReadChannel = response.body()
+    override fun streamChatMessage(messageRequest: MessageRequest): Flow<MessageResponseChunk> = flow<MessageResponseChunk> {
         try {
-            while (currentCoroutineContext().isActive && !channel.isClosedForRead) {
-                val line = channel.readUTF8Line() ?: continue
-                val value: T = when {
-                    line.startsWith(STREAM_END_TOKEN) -> break
-                    line.startsWith(STREAM_PREFIX) -> Json.decodeFromString(line.removePrefix(STREAM_PREFIX))
-                    else -> continue
+            networkClient()
+                .sse(
+                    urlString = if (apiUrl.endsWith("/")) "${apiUrl}v1/messages" else "$apiUrl/v1/messages",
+                    request = {
+                        method = HttpMethod.Post
+                        setBody(Json.encodeToJsonElement(messageRequest))
+                        accept(ContentType.Text.EventStream)
+                        headers {
+                            append(API_KEY_HEADER, token ?: "")
+                            append(VERSION_HEADER, ANTHROPIC_VERSION)
+                        }
+                    }
+                ) {
+                    incoming.collect { event -> event.data?.let { line -> emit(Json.decodeFromString(line)) } }
                 }
-                emit(value)
-            }
-        } finally {
-            channel.cancel()
+        } catch (e: Exception) {
+            emit(ErrorResponseChunk(error = ErrorDetail(type = "network_error", message = e.message ?: "")))
         }
     }
 
     companion object {
-        private const val STREAM_PREFIX = "data:"
-        private const val STREAM_END_TOKEN = "event: message_stop"
         private const val API_KEY_HEADER = "x-api-key"
         private const val VERSION_HEADER = "anthropic-version"
         private const val ANTHROPIC_VERSION = "2023-06-01"

app/src/main/kotlin/dev/chungjungsoo/gptmobile/data/network/NetworkClient.kt

@@ -9,6 +9,7 @@ import io.ktor.client.plugins.logging.DEFAULT
 import io.ktor.client.plugins.logging.LogLevel
 import io.ktor.client.plugins.logging.Logger
 import io.ktor.client.plugins.logging.Logging
+import io.ktor.client.plugins.sse.SSE
 import io.ktor.client.request.header
 import io.ktor.http.ContentType
 import io.ktor.http.HttpHeaders
@@ -37,6 +38,8 @@ class NetworkClient @Inject constructor(
                 )
             }
 
+            install(SSE)
+
             install(HttpTimeout) {
                 requestTimeoutMillis = TIMEOUT.toLong()
             }

gradle/libs.versions.toml

@@ -13,11 +13,11 @@ datastore = "1.1.1"
 gemini = "0.9.0"
 hilt = "2.52"
 ksp = "2.0.20-1.0.25" # Also change with Kotlin version
-ktor = "2.3.12"
+ktor = "3.0.1"
 androidxHilt = "1.2.0"
 navigation = "2.8.4"
 markdown = "0.5.4"
-openai = "3.8.2"
+openai = "4.0.0-beta01"
 serialization = "1.7.3" # Should not update due to kotlin version
 splashscreen = "1.0.1"
 room = "2.6.1"