Prevent propagating exception to top-level

Fix #1356 #1237
ktorio · Jan 13, 2020 · b50206c · b50206c
1 parent 6365524
commit b50206c
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 14 deletions.
diff --git a/ktor-client/ktor-client-cio/jvm/src/io/ktor/client/engine/cio/Endpoint.kt b/ktor-client/ktor-client-cio/jvm/src/io/ktor/client/engine/cio/Endpoint.kt
@@ -10,7 +10,6 @@ import io.ktor.network.sockets.Socket
 import io.ktor.network.tls.*
 import io.ktor.util.*
 import io.ktor.util.date.*
-import io.ktor.utils.io.*
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
@@ -61,11 +60,13 @@ internal class Endpoint(
         }
     }
 
-    suspend fun execute(request: HttpRequestData, callContext: CoroutineContext): HttpResponseData =
-        suspendCancellableCoroutine { continuation ->
-            val task = RequestTask(request, continuation, callContext)
-            tasks.offer(task)
-        }
+    suspend fun execute(
+        request: HttpRequestData,
+        callContext: CoroutineContext
+    ): HttpResponseData = suspendCancellableCoroutine { continuation ->
+        val task = RequestTask(request, continuation, callContext)
+        tasks.offer(task)
+    }
 
     private suspend fun makePipelineRequest(task: RequestTask) {
         if (deliveryPoint.offer(task)) return

diff --git a/ktor-client/ktor-client-core/jvm/src/io/ktor/client/utils/CoroutineDispatcherUtils.kt b/ktor-client/ktor-client-core/jvm/src/io/ktor/client/utils/CoroutineDispatcherUtils.kt
@@ -13,12 +13,17 @@ import java.util.concurrent.atomic.*
  * Creates [CoroutineDispatcher] based on thread pool of [threadCount] threads.
  */
 @InternalAPI
-fun Dispatchers.fixedThreadPoolDispatcher(threadCount: Int, threadName: String = "thread-pool-%d"): CoroutineDispatcher {
+fun Dispatchers.fixedThreadPoolDispatcher(
+    threadCount: Int,
+    threadName: String = "thread-pool-%d"
+): CoroutineDispatcher {
     val threadsNum = AtomicInteger(0)
     return Executors.newFixedThreadPool(threadCount) {
         Thread(it).apply {
             isDaemon = true
             name = threadName.format(threadsNum.getAndIncrement())
+
+            setUncaughtExceptionHandler { _, _ -> }
         }
     }.asCoroutineDispatcher()
 }
diff --git a/ktor-network/jvm/src/io/ktor/network/selector/ActorSelectorManager.kt b/ktor-network/jvm/src/io/ktor/network/selector/ActorSelectorManager.kt
@@ -7,19 +7,18 @@ package io.ktor.network.selector
 import io.ktor.util.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
-import java.io.Closeable
+import java.io.*
 import java.nio.channels.*
 import java.util.concurrent.atomic.*
 import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.*
-import kotlin.jvm.*
 
 /**
  * Default CIO selector manager implementation
  */
 @Suppress("BlockingMethodInNonBlockingContext")
 @KtorExperimentalAPI
-class ActorSelectorManager(dispatcher: CoroutineContext) : SelectorManagerSupport(), Closeable, CoroutineScope {
+class ActorSelectorManager(context: CoroutineContext) : SelectorManagerSupport(), Closeable, CoroutineScope {
     @Volatile
     private var selectorRef: Selector? = null
 
@@ -35,7 +34,7 @@ class ActorSelectorManager(dispatcher: CoroutineContext) : SelectorManagerSuppor
 
     private val mb = LockFreeMPSCQueue<Selectable>()
 
-    override val coroutineContext: CoroutineContext = dispatcher + CoroutineName("selector")
+    override val coroutineContext: CoroutineContext = context + CoroutineName("selector")
 
     init {
         launch {
@@ -143,8 +142,7 @@ class ActorSelectorManager(dispatcher: CoroutineContext) : SelectorManagerSuppor
                 if (!continuation.resume(Unit)) {
                     selectWakeup()
                 }
-            }
-            else if (selectable.channel.isOpen) throw ClosedSelectorException()
+            } else if (selectable.channel.isOpen) throw ClosedSelectorException()
             else throw ClosedChannelException()
         } catch (t: Throwable) {
             cancelAllSuspensions(selectable, t)
@@ -186,7 +184,8 @@ class ActorSelectorManager(dispatcher: CoroutineContext) : SelectorManagerSuppor
         fun resume(value: R): Boolean {
             val continuation = ref.getAndSet(null)
             if (continuation != null) {
-                continuation.resume(value) /** we resume unintercepted, see [dispatchIfNeeded] */
+                continuation.resume(value)
+                /** we resume unintercepted, see [dispatchIfNeeded] */
                 return true
             }