feat: bind okhttp engine directly (#663)

Remove Ktor as a middleman between us and OkHttp. The default engine is still OkHttp based as it was before
but now the integration is directly mapped to OkHttp rather than through Ktor.

See also #656, #662

Author: aajtodd

.changes/c6d1863b-170d-4622-8900-19a75a9ebf9d.json

@@ -0,0 +1,8 @@
+{
+    "id": "c6d1863b-170d-4622-8900-19a75a9ebf9d",
+    "type": "misc",
+    "description": "Refactor to bind directly to okhttp and remove ktor as a middleman",
+    "issues": [
+        "awslabs/smithy-kotlin#629"
+    ]
+}

build.gradle.kts

@@ -136,7 +136,7 @@ dependencies {
 val lintPaths = listOf(
     "smithy-kotlin-codegen/src/**/*.kt",
     "runtime/**/*.kt",
-    "benchmarks/**/jvm/*.kt",
+    "tests/**/jvm/**/*.kt",
 )
 
 tasks.register<JavaExec>("ktlint") {

gradle.properties

@@ -16,10 +16,11 @@ dokkaVersion=1.6.21
 
 # kotlin libraries
 coroutinesVersion=1.6.1
-ktorVersion=2.0.1
-atomicFuVersion=0.17.2
+ktorVersion=2.0.2
+atomicFuVersion=0.17.3
 kotlinxSerializationVersion=1.3.3
 jsoupVersion=1.14.3
+okHttpVersion=5.0.0-alpha.9
 
 # codegen
 smithyVersion=1.17.0
@@ -43,4 +44,4 @@ kotlinLoggingVersion=2.1.21
 slf4jVersion=1.7.36
 
 # crt
-crtKotlinVersion=0.6.0
+crtKotlinVersion=0.6.0

runtime/protocol/http-client-engines/http-client-engine-crt/common/src/aws/smithy/kotlin/runtime/http/engine/crt/SdkStreamResponseHandler.kt

@@ -74,10 +74,10 @@ internal class SdkStreamResponseHandler(
         }
     }
 
-    private fun createHttpResponseBody(contentLength: Long): HttpBody {
+    private fun createHttpResponseBody(contentLength: Long?): HttpBody {
         sdkBody = bufferedReadChannel(::onDataConsumed)
         return object : HttpBody.Streaming() {
-            override val contentLength: Long = contentLength
+            override val contentLength: Long? = contentLength
             override fun readFrom(): SdkByteReadChannel =
                 sdkBody!!
         }
@@ -90,10 +90,10 @@ internal class SdkStreamResponseHandler(
 
         val transferEncoding = headers["Transfer-Encoding"]?.lowercase()
         val chunked = transferEncoding == "chunked"
-        val contentLength = headers["Content-Length"]?.toLong() ?: 0
+        val contentLength = headers["Content-Length"]?.toLong()
         val status = HttpStatusCode.fromValue(stream.responseStatusCode)
 
-        val hasBody = (contentLength > 0 || chunked) &&
+        val hasBody = ((contentLength != null && contentLength > 0) || chunked) &&
             (status !in listOf(HttpStatusCode.NotModified, HttpStatusCode.NoContent)) &&
             !status.isInformational()
 

runtime/protocol/http-client-engines/http-client-engine-default/build.gradle.kts

@@ -18,9 +18,8 @@ kotlin {
         }
         jvmMain {
             dependencies {
-                implementation(project(":runtime:protocol:http-client-engines:http-client-engine-ktor"))
                 // okhttp works on both JVM and Android
-                implementation("io.ktor:ktor-client-okhttp:$ktorVersion")
+                implementation(project(":runtime:protocol:http-client-engines:http-client-engine-okhttp"))
             }
         }
         all {

runtime/protocol/http-client-engines/http-client-engine-default/common/src/aws/smithy/kotlin/runtime/http/engine/DefaultHttpEngine.kt

@@ -8,13 +8,6 @@ package aws.smithy.kotlin.runtime.http.engine
 /**
  * Factory function to create a new HTTP client engine using the default for the current KMP target
  */
-fun DefaultHttpEngine(config: HttpClientEngineConfig = HttpClientEngineConfig.Default): HttpClientEngine =
-    newDefaultHttpEngine(config)
+fun DefaultHttpEngine(block: (HttpClientEngineConfig.Builder.() -> Unit)? = null): HttpClientEngine = newDefaultHttpEngine(block)
 
-fun DefaultHttpEngine(block: HttpClientEngineConfig.Builder.() -> Unit): HttpClientEngine {
-    val builder = HttpClientEngineConfig.Builder().apply(block)
-    val config = HttpClientEngineConfig(builder)
-    return DefaultHttpEngine(config)
-}
-
-internal expect fun newDefaultHttpEngine(config: HttpClientEngineConfig): HttpClientEngine
+internal expect fun newDefaultHttpEngine(block: (HttpClientEngineConfig.Builder.() -> Unit)?): HttpClientEngine

runtime/protocol/http-client-engines/http-client-engine-default/jvm/src/aws/smithy/kotlin/runtime/http/engine/DefaultHttpEngineJVM.kt

@@ -5,40 +5,10 @@
 
 package aws.smithy.kotlin.runtime.http.engine
 
-import aws.smithy.kotlin.runtime.http.engine.ktor.KtorEngine
-import io.ktor.client.engine.okhttp.*
-import okhttp3.ConnectionPool
-import okhttp3.Protocol
-import java.util.concurrent.TimeUnit
-import kotlin.time.toJavaDuration
+import aws.smithy.kotlin.runtime.http.engine.okhttp.OkHttpEngine
 
-internal actual fun newDefaultHttpEngine(config: HttpClientEngineConfig): HttpClientEngine {
-    val okHttpEngine = OkHttp.create {
-        config {
-            connectTimeout(config.connectTimeout.toJavaDuration())
-            readTimeout(config.socketReadTimeout.toJavaDuration())
-            writeTimeout(config.socketWriteTimeout.toJavaDuration())
-            val pool = ConnectionPool(
-                maxIdleConnections = config.maxConnections.toInt(),
-                keepAliveDuration = config.connectionIdleTimeout.inWholeMilliseconds,
-                TimeUnit.MILLISECONDS
-            )
-            connectionPool(pool)
-
-            eventListener(HttpEngineEventListener(pool))
-
-            if (config.alpn.isNotEmpty()) {
-                val protocols = config.alpn.mapNotNull {
-                    when (it) {
-                        AlpnId.HTTP1_1 -> Protocol.HTTP_1_1
-                        AlpnId.HTTP2 -> Protocol.HTTP_2
-                        else -> null
-                    }
-                }
-                protocols(protocols)
-            }
-        }
-    }
-
-    return KtorEngine(okHttpEngine)
+internal actual fun newDefaultHttpEngine(block: (HttpClientEngineConfig.Builder.() -> Unit)?): HttpClientEngine = if (block != null) {
+    OkHttpEngine(block)
+} else {
+    OkHttpEngine()
 }

runtime/protocol/http-client-engines/http-client-engine-ktor/common/src/aws/smithy/kotlin/runtime/http/engine/ktor/KtorEngine.kt

@@ -8,18 +8,17 @@ import aws.smithy.kotlin.runtime.client.ExecutionContext
 import aws.smithy.kotlin.runtime.http.Headers
 import aws.smithy.kotlin.runtime.http.HttpStatusCode
 import aws.smithy.kotlin.runtime.http.engine.*
-import aws.smithy.kotlin.runtime.http.operation.getLogger
 import aws.smithy.kotlin.runtime.http.operation.withContext
 import aws.smithy.kotlin.runtime.http.request.HttpRequest
 import aws.smithy.kotlin.runtime.http.response.HttpCall
 import aws.smithy.kotlin.runtime.http.response.HttpResponse
 import aws.smithy.kotlin.runtime.logging.Logger
 import aws.smithy.kotlin.runtime.logging.trace
 import aws.smithy.kotlin.runtime.time.Instant
+import aws.smithy.kotlin.runtime.util.InternalApi
 import io.ktor.client.*
 import io.ktor.client.call.*
 import io.ktor.client.request.*
-import io.ktor.client.statement.*
 import io.ktor.http.*
 import io.ktor.util.*
 import io.ktor.utils.io.*
@@ -38,6 +37,7 @@ import io.ktor.client.engine.HttpClientEngine as KtorHttpClientEngine
  * This class can be used to wrap any Ktor compliant engine (though not all engines
  * may support HTTP features required by any given SDK).
  */
+@InternalApi // FIXME - decide on whether to support in GA or not. Most use cases would be better off by wrapping the underlying ktor engine directly
 class KtorEngine(
     private val engine: KtorHttpClientEngine
 ) : HttpClientEngineBase("ktor") {

runtime/protocol/http-client-engines/http-client-engine-okhttp/build.gradle.kts

@@ -0,0 +1,37 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+description = "OkHttp Client Engine for Smithy services generated by smithy-kotlin"
+extra["displayName"] = "Smithy :: Kotlin :: HTTP :: Engine :: OkHttp"
+extra["moduleName"] = "aws.smithy.kotlin.runtime.http.engine.okhttp"
+
+val coroutinesVersion: String by project
+val okHttpVersion: String by project
+
+kotlin {
+    sourceSets {
+        commonMain {
+            dependencies {
+                api(project(":runtime:protocol:http"))
+                implementation(project(":runtime:logging"))
+                implementation(project(":runtime:io"))
+
+                implementation("com.squareup.okhttp3:okhttp:$okHttpVersion")
+                implementation("com.squareup.okhttp3:okhttp-coroutines:$okHttpVersion")
+
+            }
+        }
+        commonTest {
+            dependencies {
+                implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:$coroutinesVersion")
+                implementation(project(":runtime:hashing"))
+            }
+        }
+
+        all {
+            languageSettings.optIn("aws.smithy.kotlin.runtime.util.InternalApi")
+        }
+    }
+}

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/ByteChannelRequestBody.kt

@@ -0,0 +1,58 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+package aws.smithy.kotlin.runtime.http.engine.okhttp
+
+import aws.smithy.kotlin.runtime.http.HttpBody
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.runBlocking
+import okhttp3.MediaType
+import okhttp3.RequestBody
+import okio.BufferedSink
+import java.nio.ByteBuffer
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * OkHttp [RequestBody] that reads from [body] channel
+ */
+internal class ByteChannelRequestBody(
+    private val body: HttpBody.Streaming,
+    private val callContext: CoroutineContext
+) : RequestBody() {
+    override fun contentType(): MediaType? = null
+    override fun contentLength(): Long = body.contentLength ?: -1
+    override fun isOneShot(): Boolean = !body.isReplayable
+
+    // TODO - enable for event streams. Requires different processing of request body
+    override fun isDuplex(): Boolean = false
+
+    @OptIn(ExperimentalStdlibApi::class)
+    override fun writeTo(sink: BufferedSink) {
+        // remove the current dispatcher (if it exists) and use the internal
+        // runBlocking dispatcher that blocks the current thread
+        val sendContext = callContext.minusKey(CoroutineDispatcher) + callContext.derivedName("send-request-body")
+
+        // Non-duplex (aka "normal") requests MUST write all of their request body
+        // before this function returns. Requests are given a background thread to
+        // do this work in, and it is safe and expected to block.
+        // see: https://square.github.io/okhttp/4.x/okhttp/okhttp3/-request-body/is-duplex/
+        runBlocking(sendContext) {
+            val chan = body.readFrom()
+            val buffer = ByteBuffer.allocate(DEFAULT_BUFFER_SIZE)
+            while (!chan.isClosedForRead) {
+                // fill the buffer by reading chunks from the underlying source
+                while (chan.readAvailable(buffer) != -1 && buffer.remaining() > 0) {
+                }
+
+                buffer.flip()
+                while (buffer.remaining() > 0) {
+                    sink.write(buffer)
+                }
+
+                buffer.clear()
+            }
+        }
+    }
+}