KTOR-4419 Ignore request timeout for WebSocket requests (#3062)

ktor-client/ktor-client-cio/jvmAndNix/src/io/ktor/client/engine/cio/CIOEngine.kt

@@ -14,7 +14,6 @@ import io.ktor.network.selector.*
 import io.ktor.util.*
 import io.ktor.util.collections.*
 import io.ktor.util.network.*
-import io.ktor.utils.io.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import kotlin.coroutines.*
@@ -24,7 +23,7 @@ internal class CIOEngine(
     override val config: CIOEngineConfig
 ) : HttpClientEngineBase("ktor-cio") {
 
-    override val dispatcher by lazy {
+    override val dispatcher: CoroutineDispatcher by lazy {
         Dispatchers.clientDispatcher(config.threadsCount, "ktor-cio-dispatcher")
     }
 

ktor-client/ktor-client-cio/jvmAndNix/src/io/ktor/client/engine/cio/Endpoint.kt

@@ -9,6 +9,7 @@ import io.ktor.client.network.sockets.*
 import io.ktor.client.plugins.*
 import io.ktor.client.request.*
 import io.ktor.client.utils.*
+import io.ktor.http.*
 import io.ktor.network.sockets.*
 import io.ktor.network.tls.*
 import io.ktor.util.*
@@ -114,7 +115,7 @@ internal class Endpoint(
                 }
             }
 
-            val timeout = getRequestTimeout(request.getCapabilityOrNull(HttpTimeout), config)
+            val timeout = getRequestTimeout(request, config)
             setupTimeout(callContext, request, timeout)
 
             val requestTime = GMTDate()
@@ -256,7 +257,18 @@ public class FailToConnectException : Exception("Connect timed out or retry atte
 
 internal expect fun Throwable.mapToKtor(request: HttpRequestData): Throwable
 
+@OptIn(InternalAPI::class)
 internal fun getRequestTimeout(
-    requestConfig: HttpTimeout.HttpTimeoutCapabilityConfiguration?,
+    request: HttpRequestData,
     engineConfig: CIOEngineConfig
-): Long = requestConfig?.requestTimeoutMillis ?: engineConfig.requestTimeout
+): Long {
+    /**
+     * The request timeout is handled by the plugin and disabled for the WebSockets.
+     */
+    val isWebSocket = request.url.protocol.isWebsocket()
+    if (request.getCapabilityOrNull(HttpTimeout) != null || isWebSocket || request.isUpgradeRequest()) {
+        return HttpTimeout.INFINITE_TIMEOUT_MS
+    }
+
+    return engineConfig.requestTimeout
+}

ktor-client/ktor-client-cio/jvmAndNix/test/CIOEngineTest.kt

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2014-2022 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+import io.ktor.client.*
+import io.ktor.client.engine.cio.*
+import io.ktor.client.plugins.websocket.*
+import io.ktor.client.tests.utils.*
+import io.ktor.websocket.*
+import kotlinx.coroutines.*
+import kotlin.test.*
+
+class CIOEngineTest {
+
+    @Test
+    fun testRequestTimeoutIgnoredWithWebSocket(): Unit = runBlocking {
+        val client = HttpClient(CIO) {
+            engine {
+                requestTimeout = 10
+            }
+
+            install(WebSockets)
+        }
+
+        var received = false
+        client.ws("$TEST_WEBSOCKET_SERVER/websockets/echo") {
+            delay(20)
+
+            send(Frame.Text("Hello"))
+
+            val response = incoming.receive() as Frame.Text
+            received = true
+            assertEquals("Hello", response.readText())
+        }
+
+        assertTrue(received)
+    }
+}

ktor-client/ktor-client-core/common/src/io/ktor/client/plugins/HttpTimeout.kt

@@ -9,6 +9,7 @@ import io.ktor.client.engine.*
 import io.ktor.client.network.sockets.*
 import io.ktor.client.request.*
 import io.ktor.client.utils.*
+import io.ktor.http.*
 import io.ktor.util.*
 import io.ktor.utils.io.errors.*
 import kotlinx.coroutines.*
@@ -135,8 +136,12 @@ public class HttpTimeout private constructor(
         override fun prepare(block: HttpTimeoutCapabilityConfiguration.() -> Unit): HttpTimeout =
             HttpTimeoutCapabilityConfiguration().apply(block).build()
 
+        @OptIn(InternalAPI::class)
         override fun install(plugin: HttpTimeout, scope: HttpClient) {
             scope.requestPipeline.intercept(HttpRequestPipeline.Before) {
+                val isWebSocket = context.url.protocol.isWebsocket()
+                if (isWebSocket || context.body is ClientUpgradeContent) return@intercept
+
                 var configuration = context.getCapabilityOrNull(HttpTimeout)
                 if (configuration == null && plugin.hasNotNullTimeouts()) {
                     configuration = HttpTimeoutCapabilityConfiguration()

ktor-client/ktor-client-tests/common/test/io/ktor/client/tests/WebSocketTest.kt

@@ -14,11 +14,15 @@ import io.ktor.test.dispatcher.*
 import io.ktor.util.reflect.*
 import io.ktor.utils.io.charsets.*
 import io.ktor.websocket.*
+import kotlinx.coroutines.*
 import kotlin.test.*
 
+internal val ENGINES_WITHOUT_WS = listOf("Android", "Apache", "Curl")
+
 private const val TEST_SIZE: Int = 100
 
 class WebSocketTest : ClientLoader() {
+
     data class Data(val stringValue: String)
 
     private val customContentConverter = object : WebsocketContentConverter {
@@ -62,7 +66,7 @@ class WebSocketTest : ClientLoader() {
     }
 
     @Test
-    fun testWebsocketSession() = clientTests(listOf("Android", "Apache", "Curl")) {
+    fun testWebsocketSession() = clientTests(ENGINES_WITHOUT_WS) {
         config {
             install(WebSockets)
         }
@@ -77,7 +81,7 @@ class WebSocketTest : ClientLoader() {
     }
 
     @Test
-    fun testWebsocketWithDefaultRequest() = clientTests(listOf("Android", "Apache", "Curl")) {
+    fun testWebsocketWithDefaultRequest() = clientTests(ENGINES_WITHOUT_WS) {
         config {
             install(WebSockets)
             defaultRequest {
@@ -97,7 +101,7 @@ class WebSocketTest : ClientLoader() {
     }
 
     @Test
-    fun testWebsocketSessionWithError() = clientTests(listOf("Android", "Apache", "Curl")) {
+    fun testWebsocketSessionWithError() = clientTests(ENGINES_WITHOUT_WS) {
         config {
             install(WebSockets)
         }
@@ -123,7 +127,7 @@ class WebSocketTest : ClientLoader() {
     }
 
     @Test
-    fun testWebSocketSerialization() = clientTests(listOf("Android", "Apache", "Curl")) {
+    fun testWebSocketSerialization() = clientTests(ENGINES_WITHOUT_WS) {
         config {
             WebSockets {
                 contentConverter = customContentConverter
@@ -143,7 +147,7 @@ class WebSocketTest : ClientLoader() {
     }
 
     @Test
-    fun testSerializationWithNoConverter() = clientTests(listOf("Android", "Apache", "Curl")) {
+    fun testSerializationWithNoConverter() = clientTests(ENGINES_WITHOUT_WS) {
         config {
             WebSockets {
             }
@@ -165,7 +169,7 @@ class WebSocketTest : ClientLoader() {
     }
 
     @Test
-    fun testQueryParameters() = clientTests(listOf("Android", "Apache", "Curl")) {
+    fun testQueryParameters() = clientTests(ENGINES_WITHOUT_WS) {
         config {
             install(WebSockets)
         }
@@ -182,4 +186,25 @@ class WebSocketTest : ClientLoader() {
             }
         }
     }
+
+    @Test
+    fun testRequestTimeoutIsNotApplied() = clientTests(ENGINES_WITHOUT_WS) {
+        config {
+            install(WebSockets)
+
+            install(HttpTimeout) {
+                requestTimeoutMillis = 10
+            }
+        }
+
+        test { client ->
+            client.webSocket("$TEST_WEBSOCKET_SERVER/websockets/echo") {
+                delay(20)
+
+                send("test")
+                val result = incoming.receive() as Frame.Text
+                assertEquals("test", result.readText())
+            }
+        }
+    }
 }