Merge branch 'release/3.x'

Author: osipxd

.idea/vcs.xml

@@ -1,6 +1,6 @@
 /*
-* Copyright 2014-2021 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
-*/
+ * Copyright 2014-2026 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
 
 package io.ktor.client.plugins
 
@@ -10,12 +10,11 @@ import io.ktor.client.plugins.api.*
 import io.ktor.client.request.*
 import io.ktor.http.*
 import io.ktor.http.content.*
-import io.ktor.util.*
 import io.ktor.util.logging.*
 import io.ktor.utils.io.*
 import io.ktor.utils.io.charsets.*
 import io.ktor.utils.io.core.*
-import kotlin.math.*
+import kotlin.math.roundToInt
 
 private val LOGGER = KtorSimpleLogger("io.ktor.client.plugins.HttpPlainText")
 
@@ -68,7 +67,7 @@ public class HttpPlainTextConfig {
  * [HttpClient] plugin that encodes [String] request bodies to [TextContent]
  * and processes the response body as [String].
  *
- * To configure charsets set following properties in [HttpPlainText.Config].
+ * To configure charsets, use properties in [HttpPlainTextConfig].
  *
  * [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.plugins.HttpPlainText)
  */
@@ -81,24 +80,28 @@ public val HttpPlainText: ClientPlugin<HttpPlainTextConfig> =
             .filter { !pluginConfig.charsetQuality.containsKey(it) }
             .sortedBy { it.name }
 
-        val acceptCharsetHeader = buildString {
-            withoutQuality.forEach {
-                if (isNotEmpty()) append(",")
-                append(it.name)
-            }
-
-            withQuality.forEach { (charset, quality) ->
-                if (isNotEmpty()) append(",")
-
-                check(quality in 0.0..1.0)
-
-                val truncatedQuality = (100 * quality).roundToInt() / 100.0
-                append("${charset.name};q=$truncatedQuality")
-            }
-
-            if (isEmpty()) {
-                append(responseCharsetFallback.name)
+        // RFC-9110 assumes UTF-8 as the default encoding. The Accept-Charset header is deprecated,
+        // so we shouldn't add it if the user didn't specify any charset configuration
+        val hasRegisteredCharsets = pluginConfig.charsets.any { it != Charsets.UTF_8 } ||
+            pluginConfig.charsetQuality.keys.any { it != Charsets.UTF_8 }
+        val acceptCharsetHeader = if (hasRegisteredCharsets) {
+            buildString {
+                withoutQuality.forEach {
+                    if (isNotEmpty()) append(",")
+                    append(it.name)
+                }
+
+                withQuality.forEach { (charset, quality) ->
+                    if (isNotEmpty()) append(",")
+
+                    check(quality in 0.0..1.0)
+
+                    val truncatedQuality = (100 * quality).roundToInt() / 100.0
+                    append("${charset.name};q=$truncatedQuality")
+                }
             }
+        } else {
+            null
         }
 
         val requestCharset = pluginConfig.sendCharset
@@ -122,14 +125,14 @@ public val HttpPlainText: ClientPlugin<HttpPlainTextConfig> =
             return body.readText(charset = actualCharset)
         }
 
-        fun addCharsetHeaders(context: HttpRequestBuilder) {
-            if (context.headers[HttpHeaders.AcceptCharset] != null) return
-            LOGGER.trace("Adding Accept-Charset=$acceptCharsetHeader to ${context.url}")
-            context.headers[HttpHeaders.AcceptCharset] = acceptCharsetHeader
+        fun HttpRequestBuilder.addAcceptCharsetHeader(value: String?) {
+            if (value == null || headers[HttpHeaders.AcceptCharset] != null) return
+            LOGGER.trace("Adding Accept-Charset=$value to $url")
+            headers[HttpHeaders.AcceptCharset] = value
         }
 
         on(RenderRequestHook) { request, content ->
-            addCharsetHeaders(request)
+            request.addAcceptCharsetHeader(acceptCharsetHeader)
 
             if (content !is String) return@on null
 

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

@@ -1,11 +1,13 @@
 /*
-* Copyright 2014-2021 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
-*/
+ * Copyright 2014-2026 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
 
 package io.ktor.client.utils
 
 import io.ktor.utils.io.*
-import kotlinx.coroutines.*
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
 
 /**
  * Creates [CoroutineDispatcher] for the client with fixed [threadCount] and specified [dispatcherName].
@@ -16,6 +18,14 @@ import kotlinx.coroutines.*
  * @param threadCount the number of threads for the new [CoroutineDispatcher].
  * @param dispatcherName the name of the new [CoroutineDispatcher].
  */
+@OptIn(DelicateCoroutinesApi::class)
+@Deprecated(
+    "Use Dispatchers.IO.limitedParallelism instead",
+    ReplaceWith(
+        "Dispatchers.IO.limitedParallelism(threadCount)",
+        "kotlinx.coroutines.Dispatchers",
+    )
+)
 @InternalAPI
 public expect fun Dispatchers.clientDispatcher(
     threadCount: Int,

ktor-client/ktor-client-core/common/src/io/ktor/client/utils/CoroutineUtils.kt

@@ -1,20 +1,23 @@
 /*
- * Copyright 2014-2019 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2014-2026 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package io.ktor.client.utils
 
-import io.ktor.util.*
 import io.ktor.utils.io.*
-import kotlinx.coroutines.*
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.newFixedThreadPoolContext
 
 /**
  * Creates [CoroutineDispatcher] for client with fixed [threadCount] and specified [dispatcherName].
  *
  * [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.utils.clientDispatcher)
  */
+@OptIn(DelicateCoroutinesApi::class)
 @InternalAPI
 public actual fun Dispatchers.clientDispatcher(
     threadCount: Int,
     dispatcherName: String
-): CoroutineDispatcher = Dispatchers.createFixedThreadDispatcher(dispatcherName, threadCount)
+): CoroutineDispatcher = newFixedThreadPoolContext(threadCount, dispatcherName)

ktor-client/ktor-client-core/posix/src/io/ktor/client/utils/CoroutineUtilsPosix.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2014-2026 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package io.ktor.client.engine.curl
@@ -51,7 +51,12 @@ internal class CurlClientEngine(
             val responseBody: Any = if (data.isUpgradeRequest()) {
                 val wsConfig = data.attributes[WEBSOCKETS_KEY]
                 val websocket = responseBody as CurlWebSocketResponseBody
-                CurlWebSocketSession(websocket, callContext, wsConfig.channelsConfig.outgoing)
+                CurlWebSocketSession(
+                    websocket,
+                    callContext,
+                    wsConfig.channelsConfig.outgoing,
+                    curlProcessor,
+                )
             } else {
                 val httpResponse = responseBody as CurlHttpResponseBody
                 data.attributes.getOrNull(ResponseAdapterAttributeKey)

ktor-client/ktor-client-curl/desktop/src/io/ktor/client/engine/curl/CurlClientEngine.kt

@@ -1,12 +1,11 @@
 /*
- * Copyright 2014-2019 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2014-2026 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package io.ktor.client.engine.curl
 
+import io.ktor.client.engine.curl.CurlTask.*
 import io.ktor.client.engine.curl.internal.*
-import io.ktor.util.*
-import io.ktor.utils.io.*
 import kotlinx.atomicfu.atomic
 import kotlinx.cinterop.ExperimentalForeignApi
 import kotlinx.cinterop.IntVar
@@ -17,21 +16,17 @@ import kotlinx.coroutines.channels.Channel
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.cancellation.CancellationException
 
-internal class RequestContainer(
-    val requestData: CurlRequestData,
-    val completionHandler: CompletableDeferred<CurlSuccess>
-)
-
+@OptIn(ExperimentalForeignApi::class)
 internal class CurlProcessor(coroutineContext: CoroutineContext) {
-    @OptIn(InternalAPI::class)
-    private val curlDispatcher: CloseableCoroutineDispatcher =
-        Dispatchers.createFixedThreadDispatcher("curl-dispatcher", 1)
+
+    @OptIn(DelicateCoroutinesApi::class, ExperimentalCoroutinesApi::class)
+    private val curlDispatcher = newSingleThreadContext("curl-dispatcher")
 
     private var curlApi: CurlMultiApiHandler? by atomic(null)
     private val closed = atomic(false)
 
     private val curlScope = CoroutineScope(coroutineContext + curlDispatcher)
-    private val requestQueue: Channel<RequestContainer> = Channel(Channel.UNLIMITED)
+    private val taskQueue: Channel<CurlTask> = Channel(Channel.UNLIMITED)
 
     init {
         val init = curlScope.launch {
@@ -49,50 +44,65 @@ internal class CurlProcessor(coroutineContext: CoroutineContext) {
 
     suspend fun executeRequest(request: CurlRequestData): CurlSuccess {
         val result = CompletableDeferred<CurlSuccess>()
-        requestQueue.send(RequestContainer(request, result))
+        taskQueue.send(SendRequest(request, result))
         curlApi!!.wakeup()
         return result.await()
     }
 
-    @OptIn(DelicateCoroutinesApi::class, ExperimentalForeignApi::class)
-    private fun runEventLoop(): Job = curlScope.launch {
+    suspend fun sendWebSocketFrame(websocket: CurlWebSocketResponseBody, flags: Int, data: ByteArray) {
+        val result = Job()
+        taskQueue.send(SendWebSocketFrame(websocket, flags, data, result))
+        curlApi!!.wakeup()
+        result.join()
+    }
+
+    @OptIn(DelicateCoroutinesApi::class)
+    private fun runEventLoop(): Job = curlScope.launch(CoroutineName("curl-processor-loop")) {
         memScoped {
             val transfersRunning = alloc<IntVar>()
             val api = curlApi!!
-            while (!requestQueue.isClosedForReceive) {
-                drainRequestQueue(api)
+            while (!taskQueue.isClosedForReceive) {
+                drainTaskQueue(api)
                 api.perform(transfersRunning)
             }
         }
     }
 
-    @OptIn(ExperimentalForeignApi::class)
-    private suspend fun drainRequestQueue(api: CurlMultiApiHandler) {
+    private suspend fun drainTaskQueue(api: CurlMultiApiHandler) {
         while (true) {
-            val container = if (api.hasHandlers()) {
-                requestQueue.tryReceive()
+            val task = if (api.hasHandlers()) {
+                taskQueue.tryReceive()
             } else {
-                requestQueue.receiveCatching()
+                taskQueue.receiveCatching()
             }.getOrNull() ?: break
 
-            val requestHandler = api.scheduleRequest(container.requestData, container.completionHandler)
-
-            val requestCleaner = container.requestData.executionContext.invokeOnCompletion { cause ->
-                if (cause == null) return@invokeOnCompletion
-                cancelRequest(requestHandler, cause)
+            when (task) {
+                is SendRequest -> handleSendRequest(api, task)
+                is SendWebSocketFrame ->
+                    api.sendWebSocketFrame(task.websocket, task.flags, task.data, task.completionHandler)
             }
+        }
+    }
 
-            container.completionHandler.invokeOnCompletion {
-                requestCleaner.dispose()
-            }
+    private fun handleSendRequest(api: CurlMultiApiHandler, task: SendRequest) {
+        val (requestData, completionHandler) = task
+        val requestHandler = api.scheduleRequest(requestData, completionHandler)
+
+        val requestCleaner = requestData.executionContext.invokeOnCompletion { cause ->
+            if (cause == null) return@invokeOnCompletion
+            cancelRequest(requestHandler, cause)
+        }
+
+        completionHandler.invokeOnCompletion {
+            requestCleaner.dispose()
         }
     }
 
     @OptIn(DelicateCoroutinesApi::class)
     fun close() {
         if (!closed.compareAndSet(false, true)) return
 
-        requestQueue.close()
+        taskQueue.close()
         curlApi!!.wakeup()
 
         GlobalScope.launch(curlDispatcher) {
@@ -103,10 +113,24 @@ internal class CurlProcessor(coroutineContext: CoroutineContext) {
         }
     }
 
-    @OptIn(ExperimentalForeignApi::class)
     private fun cancelRequest(easyHandle: EasyHandle, cause: Throwable) {
         curlScope.launch {
             curlApi!!.cancelRequest(easyHandle, cause)
         }
     }
 }
+
+private sealed interface CurlTask {
+
+    data class SendRequest(
+        val requestData: CurlRequestData,
+        val completionHandler: CompletableDeferred<CurlSuccess>,
+    ) : CurlTask
+
+    class SendWebSocketFrame(
+        val websocket: CurlWebSocketResponseBody,
+        val flags: Int,
+        val data: ByteArray,
+        val completionHandler: CompletableJob,
+    ) : CurlTask
+}