Support Ktor Engine (#5142)

* add common KtorHttpEngine

* add ktor engine dependencies

* add KtorHttpEngine test

* upgradle ktor version to 2.3.2

* upgradle kotlinx-coroutines version to 1.7.3

* rename KtorEngine.kt to KtorHttpEngine.kt

* add KtorWebSocketEngine

* add KtorExtensions

* rename parameter

* move ktor engine implements to apollo-engine-ktor module

* use latest ktor version

* generate apollo-engine-ktor api

* add apollo-engine-ktor module def in libraries.toml

* add integration tests for ktor engine

* Minor tweaks in build.gradle.kts and gradle.properties

* Mark new APIs as @ApolloExperimental

* Duplicate a WebSocket tests to use the Ktor engine, and tweak KtorWebSocketEngine to resolve issues

---------

Co-authored-by: BoD <[email protected]>

Author: yt8492

gradle/libraries.toml

@@ -30,11 +30,11 @@ kotlin-plugin-min = "1.6.0"
 kotlin-plugin = "1.9.0"
 kotlin-plugin-max = "1.9.0"
 kotlin-stdlib = "1.6.21"
-kotlinx-coroutines = "1.6.4"
+kotlinx-coroutines = "1.7.3"
 kotlinx-datetime = "0.4.0"
 kotlinx-serialization-runtime = "1.5.0"
 ksp = "1.9.0-1.0.11"
-ktor = "2.2.2"
+ktor = "2.3.3"
 okhttp = "4.11.0"
 rx-android = "2.0.1"
 rx-java2 = "2.2.21"
@@ -77,6 +77,7 @@ apollo-runtime = { group = "com.apollographql.apollo3", name = "apollo-runtime",
 apollo-runtime-published = { group = "com.apollographql.apollo3", name = "apollo-runtime", version.ref = "apollo-published" }
 apollo-annotations = { group = "com.apollographql.apollo3", name = "apollo-annotations", version.ref = "apollo" }
 apollo-runtime-java = { group = "com.apollographql.apollo3", name = "apollo-runtime-java", version.ref = "apollo" }
+apollo-engine-ktor = { group = "com.apollographql.apollo3", name = "apollo-engine-ktor", version.ref = "apollo" }
 apollo-rx2 = { group = "com.apollographql.apollo3", name = "apollo-rx2-support", version.ref = "apollo" }
 apollo-rx3-java = { group = "com.apollographql.apollo3", name = "apollo-rx3-support-java", version.ref = "apollo" }
 apollo-testingsupport = { group = "com.apollographql.apollo3", name = "apollo-testing-support", version.ref = "apollo" }
@@ -144,7 +145,11 @@ kotlinx-serialization-json = { group = "org.jetbrains.kotlinx", name = "kotlinx-
 kotlinx-serialization-json-okio = { group = "org.jetbrains.kotlinx", name = "kotlinx-serialization-json-okio", version.ref = "kotlinx-serialization-runtime" }
 kotlinx-binarycompatibilityvalidator = { group = "org.jetbrains.kotlinx", name = "binary-compatibility-validator", version = "0.13.2" }
 ksp = { group = "com.google.devtools.ksp", name = "symbol-processing-gradle-plugin", version.ref = "ksp" }
+ktor-client-core = { group = "io.ktor", name= "ktor-client-core", version.ref = "ktor" }
+ktor-client-okhttp = { group = "io.ktor", name = "ktor-client-okhttp", version.ref = "ktor" }
+ktor-client-darwin = { group = "io.ktor", name = "ktor-client-darwin", version.ref = "ktor" }
 ktor-client-js = { group = "io.ktor", name = "ktor-client-js", version.ref = "ktor" }
+ktor-client-websockets = { group = "io.ktor", name = "ktor-client-websockets", version.ref = "ktor" }
 okhttp = { group = "com.squareup.okhttp3", name = "okhttp", version.ref = "okhttp" }
 okhttp-logging = { group = "com.squareup.okhttp3", name = "logging-interceptor", version.ref = "okhttp" }
 okhttp-mockwebserver = { group = "com.squareup.okhttp3", name = "mockwebserver", version.ref = "okhttp" }

libraries/apollo-engine-ktor/api/apollo-engine-ktor.api

@@ -0,0 +1,3 @@
+public final class com/apollographql/apollo3/network/KtorExtensionsKt {
+}
+

libraries/apollo-engine-ktor/build.gradle.kts

@@ -0,0 +1,49 @@
+plugins {
+  id("org.jetbrains.kotlin.multiplatform")
+  id("apollo.library")
+}
+
+apolloLibrary {
+  javaModuleName("com.apollographql.apollo3.engine.ktor")
+  mpp {
+    withLinux.set(false)
+  }
+}
+
+kotlin {
+  sourceSets {
+    findByName("commonMain")?.apply {
+      dependencies {
+        api(project(":apollo-runtime"))
+        api(libs.ktor.client.core)
+        api(libs.ktor.client.websockets)
+      }
+    }
+
+    findByName("commonTest")?.apply {
+      dependencies {
+        implementation(project(":apollo-mockserver"))
+        implementation(project(":apollo-testing-support")) {
+          because("runTest")
+          // We have a circular dependency here that creates a warning in JS
+          // w: duplicate library name: com.apollographql.apollo3:apollo-mockserver
+          // See https://youtrack.jetbrains.com/issue/KT-51110
+          // We should probably remove this circular dependency but for the time being, just use excludes
+          exclude(group =  "com.apollographql.apollo3", module = "apollo-runtime")
+        }
+      }
+    }
+
+    findByName("jvmMain")?.apply {
+      dependencies {
+        api(libs.ktor.client.okhttp)
+      }
+    }
+
+    findByName("appleMain")?.apply {
+      dependencies {
+        api(libs.ktor.client.darwin)
+      }
+    }
+  }
+}

libraries/apollo-engine-ktor/gradle.properties

@@ -0,0 +1,3 @@
+POM_ARTIFACT_ID=apollo-engine-ktor
+POM_NAME=Apollo Engine Ktor
+POM_DESCRIPTION=Http and WebSocket engines for Apollo Kotlin based on Ktor

libraries/apollo-engine-ktor/src/commonMain/kotlin/com/apollographql/apollo3/network/KtorExtensions.kt

@@ -0,0 +1,19 @@
+package com.apollographql.apollo3.network
+
+import com.apollographql.apollo3.ApolloClient
+import com.apollographql.apollo3.annotations.ApolloExperimental
+import com.apollographql.apollo3.network.http.KtorHttpEngine
+import com.apollographql.apollo3.network.ws.KtorWebSocketEngine
+import io.ktor.client.HttpClient
+
+/**
+ * Configures the [ApolloClient] to use the Ktor [HttpClient] for network requests.
+ * The [HttpClient] will be used for both HTTP and WebSocket requests.
+ *
+ * See also [ApolloClient.Builder.httpEngine] and [ApolloClient.Builder.webSocketEngine]
+ */
+@ApolloExperimental
+fun ApolloClient.Builder.ktorClient(httpClient: HttpClient) = apply {
+  httpEngine(KtorHttpEngine(httpClient))
+  webSocketEngine(KtorWebSocketEngine(httpClient))
+}

libraries/apollo-engine-ktor/src/commonMain/kotlin/com/apollographql/apollo3/network/http/KtorHttpEngine.kt

@@ -0,0 +1,77 @@
+package com.apollographql.apollo3.network.http
+
+import com.apollographql.apollo3.annotations.ApolloExperimental
+import com.apollographql.apollo3.api.http.HttpHeader
+import com.apollographql.apollo3.api.http.HttpMethod
+import com.apollographql.apollo3.api.http.HttpRequest
+import com.apollographql.apollo3.api.http.HttpResponse
+import com.apollographql.apollo3.exception.ApolloNetworkException
+import io.ktor.client.HttpClient
+import io.ktor.client.call.body
+import io.ktor.client.plugins.HttpTimeout
+import io.ktor.client.request.header
+import io.ktor.client.request.request
+import io.ktor.client.request.setBody
+import io.ktor.http.HttpHeaders
+import io.ktor.util.flattenEntries
+import okio.Buffer
+import kotlin.coroutines.cancellation.CancellationException
+
+@ApolloExperimental
+class KtorHttpEngine(
+    private val client: HttpClient,
+): HttpEngine {
+
+  private var disposed = false
+
+  constructor(timeoutMillis: Long = 60_000) : this(timeoutMillis, timeoutMillis)
+
+  constructor(connectTimeoutMillis: Long, requestTimeoutMillis: Long) : this(
+      HttpClient {
+        expectSuccess = false
+        install(HttpTimeout) {
+          this.connectTimeoutMillis = connectTimeoutMillis
+          this.requestTimeoutMillis = requestTimeoutMillis
+        }
+      }
+  )
+
+  override suspend fun execute(request: HttpRequest): HttpResponse {
+    try {
+      val response = client.request(request.url) {
+        method = when (request.method) {
+          HttpMethod.Get -> io.ktor.http.HttpMethod.Get
+          HttpMethod.Post -> io.ktor.http.HttpMethod.Post
+        }
+        request.headers.forEach {
+          header(it.name, it.value)
+        }
+        request.body?.let {
+          header(HttpHeaders.ContentType, it.contentType)
+          val buffer = Buffer()
+          it.writeTo(buffer)
+          setBody(buffer.readUtf8())
+        }
+      }
+      val responseByteArray: ByteArray = response.body()
+      val responseBufferedSource = Buffer().write(responseByteArray)
+      return HttpResponse.Builder(statusCode = response.status.value)
+          .body(responseBufferedSource)
+          .addHeaders(response.headers.flattenEntries().map { HttpHeader(it.first, it.second) })
+          .build()
+    } catch (e: CancellationException) {
+      // Cancellation Exception is passthrough
+      throw e
+    } catch (t: Throwable) {
+      throw ApolloNetworkException(t.message, t)
+    }
+  }
+
+  override fun dispose() {
+    if (!disposed) {
+      client.close()
+      disposed = true
+    }
+  }
+}
+

libraries/apollo-engine-ktor/src/commonMain/kotlin/com/apollographql/apollo3/network/ws/KtorWebSocketEngine.kt

@@ -0,0 +1,137 @@
+package com.apollographql.apollo3.network.ws
+
+import com.apollographql.apollo3.annotations.ApolloExperimental
+import com.apollographql.apollo3.api.http.HttpHeader
+import com.apollographql.apollo3.exception.ApolloWebSocketClosedException
+import io.ktor.client.HttpClient
+import io.ktor.client.plugins.websocket.WebSockets
+import io.ktor.client.plugins.websocket.webSocket
+import io.ktor.client.request.headers
+import io.ktor.client.request.url
+import io.ktor.http.URLBuilder
+import io.ktor.http.URLProtocol
+import io.ktor.http.Url
+import io.ktor.websocket.Frame
+import io.ktor.websocket.close
+import io.ktor.websocket.readText
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ClosedReceiveChannelException
+import kotlinx.coroutines.launch
+import okio.ByteString
+
+@ApolloExperimental
+class KtorWebSocketEngine(
+    private val client: HttpClient,
+) : WebSocketEngine {
+
+  constructor() : this(
+      HttpClient {
+        install(WebSockets)
+      }
+  )
+
+  private val coroutineScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+
+  override suspend fun open(
+      url: String,
+      headers: List<HttpHeader>,
+  ): WebSocketConnection = open(Url(url), headers)
+
+  private suspend fun open(url: Url, headers: List<HttpHeader>): WebSocketConnection {
+    val newUrl = URLBuilder(url).apply {
+      protocol = when (url.protocol) {
+        URLProtocol.HTTPS -> URLProtocol.WSS
+        URLProtocol.HTTP -> URLProtocol.WS
+        URLProtocol.WS, URLProtocol.WSS -> url.protocol
+        /* URLProtocol.SOCKS */else -> throw UnsupportedOperationException("SOCKS is not a supported protocol")
+      }
+    }.build()
+    val receiveMessageChannel = Channel<String>(Channel.UNLIMITED)
+    val sendFrameChannel = Channel<Frame>(Channel.UNLIMITED)
+    coroutineScope.launch {
+      try {
+        client.webSocket(
+            request = {
+              headers {
+                headers.forEach {
+                  append(it.name, it.value)
+                }
+              }
+              url(newUrl)
+            },
+        ) {
+          launch {
+            while (true) {
+              val frame = sendFrameChannel.receive()
+              try {
+                send(frame)
+              } catch (e: Exception) {
+                val closeReason = try {closeReason.await()} catch (e: Exception) {null}
+                receiveMessageChannel.close(ApolloWebSocketClosedException(code = closeReason?.code?.toInt()
+                    ?: -1, reason = closeReason?.message, cause = e))
+                sendFrameChannel.close(e)
+                break
+              }
+            }
+          }
+          while (true) {
+            when (val frame = try {
+              incoming.receive()
+            } catch (e: ClosedReceiveChannelException) {
+              val closeReason = try {closeReason.await()} catch (e: Exception) {null}
+              receiveMessageChannel.close(ApolloWebSocketClosedException(code = closeReason?.code?.toInt()
+                  ?: -1, reason = closeReason?.message, cause = e))
+              sendFrameChannel.close(e)
+              break
+            }) {
+              is Frame.Text -> {
+                receiveMessageChannel.send(frame.readText())
+              }
+
+              is Frame.Binary -> {
+                receiveMessageChannel.send(frame.data.decodeToString())
+              }
+
+              is Frame.Ping -> {
+                send(Frame.Pong(frame.data))
+              }
+
+              is Frame.Pong -> {}
+              is Frame.Close -> {
+                close()
+                receiveMessageChannel.close()
+              }
+
+              else -> error("unknown frame type")
+            }
+          }
+        }
+      } catch (e: Exception) {
+        receiveMessageChannel.close(e)
+        sendFrameChannel.close(e)
+      }
+    }
+    return object : WebSocketConnection {
+      override suspend fun receive(): String {
+        return receiveMessageChannel.receive()
+      }
+
+      override fun send(data: ByteString) {
+        sendFrameChannel.trySend(Frame.Binary(true, data.toByteArray()))
+      }
+
+      override fun send(string: String) {
+        sendFrameChannel.trySend(Frame.Text(string))
+      }
+
+      override fun close() {
+        sendFrameChannel.trySend(Frame.Close())
+        sendFrameChannel.close()
+      }
+
+    }
+  }
+}

libraries/apollo-engine-ktor/src/commonTest/kotlin/KtorHttpEngineTest.kt

@@ -0,0 +1,52 @@
+import com.apollographql.apollo3.mockserver.MockResponse
+import com.apollographql.apollo3.mockserver.MockServer
+import com.apollographql.apollo3.network.http.KtorHttpEngine
+import com.apollographql.apollo3.network.http.get
+import com.apollographql.apollo3.testing.internal.runTest
+import okio.Buffer
+import okio.ByteString
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class KtorHttpEngineTest {
+  // "Hello World" gzipped and hex encoded
+  val gzipData = """
+1f8b 0800 0000 0000 0003 f348 cdc9 c957
+08cf 2fca 4901 0056 b117 4a0b 0000 00   
+  """.replace(Regex("\\s"), "")
+
+  private fun String.toByteString(): ByteString {
+    val buffer = Buffer()
+    chunked(2).forEach {
+      buffer.writeByte(it.toInt(16))
+    }
+
+    return buffer.readByteString()
+  }
+
+  @Test
+  fun ktorEngineGzipTest() = runTest {
+    val mockServer = MockServer()
+
+    try {
+      mockServer.enqueue(MockResponse.Builder()
+          .addHeader("content-type", "application/text")
+          .addHeader("content-encoding", "gzip")
+          .body(gzipData.toByteString())
+          .build())
+
+      val engine = KtorHttpEngine()
+
+      val response = engine.get(mockServer.url())
+          .execute()
+
+      val result = response.body?.readUtf8()
+      assertEquals("Hello World", result)
+
+    } catch (e: Exception) {
+      e.printStackTrace()
+    }
+
+    mockServer.stop()
+  }
+}

libraries/apollo-testing-support/src/commonMain/kotlin/com/apollographql/apollo3/testing/runTest.kt

@@ -13,11 +13,15 @@ class MockServerTest(val mockServer: MockServer, val apolloClient: ApolloClient,
  * A convenience function that makes sure the MockServer and ApolloClient are properly closed at the end of the test
  */
 @ApolloExperimental
-fun mockServerTest(block: suspend MockServerTest.() -> Unit) = com.apollographql.apollo3.testing.internal.runTest(true) {
+fun mockServerTest(
+    clientBuilder: ApolloClient.Builder.() -> Unit = {},
+    block: suspend MockServerTest.() -> Unit
+) = com.apollographql.apollo3.testing.internal.runTest(true) {
   val mockServer = MockServer()
 
   val apolloClient = ApolloClient.Builder()
       .serverUrl(mockServer.url())
+      .apply(clientBuilder)
       .build()
 
   try {