Refactored ThreadPoolDispatcher to reuse code of ExecutorCoroutineDispatcher

Author: elizarov

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Executors.kt

@@ -37,45 +37,45 @@ public fun Executor.toCoroutineDispatcher(): CoroutineDispatcher =
 public fun Executor.asCoroutineDispatcher(): CoroutineDispatcher =
     ExecutorCoroutineDispatcher(this)
 
-private class ExecutorCoroutineDispatcher(
-    private val executor: Executor
-) : CoroutineDispatcher(), Delay {
+private class ExecutorCoroutineDispatcher(override val executor: Executor) : ExecutorCoroutineDispatcherBase()
+
+internal abstract class ExecutorCoroutineDispatcherBase : CoroutineDispatcher(), Delay {
+    abstract val executor: Executor
+
     override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
 
     override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
-        val timeout = if (executor is ScheduledExecutorService)
-            executor.schedule(ResumeUndispatchedRunnable(this, continuation), time, unit) else
-            scheduledExecutor.schedule(ResumeRunnable(continuation), time, unit)
+        val timeout = (executor as? ScheduledExecutorService)
+            ?.schedule(ResumeUndispatchedRunnable(this, continuation), time, unit)
+            ?: scheduledExecutor.schedule(ResumeRunnable(continuation), time, unit)
         continuation.cancelFutureOnCompletion(timeout)
     }
 
     override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
-        val timeout = if (executor is ScheduledExecutorService)
-            executor.schedule(block, time, unit) else
-            scheduledExecutor.schedule(block, time, unit)
+        val timeout = (executor as? ScheduledExecutorService)
+            ?.schedule(block, time, unit)
+            ?: scheduledExecutor.schedule(block, time, unit)
         return DisposableFutureHandle(timeout)
     }
 
     override fun toString(): String = executor.toString()
-    override fun equals(other: Any?): Boolean = other is ExecutorCoroutineDispatcher && other.executor === executor
+    override fun equals(other: Any?): Boolean = other is ExecutorCoroutineDispatcherBase && other.executor === executor
     override fun hashCode(): Int = System.identityHashCode(executor)
 }
 
-// --- reusing these classes in other places ---
-
-internal class ResumeUndispatchedRunnable(
-    private val dispatcher: CoroutineDispatcher,
-    private val continuation: CancellableContinuation<Unit>
+internal class ResumeRunnable(
+    private val continuation: Continuation<Unit>
 ) : Runnable {
     override fun run() {
-        with(continuation) { dispatcher.resumeUndispatched(Unit) }
+        continuation.resume(Unit)
     }
 }
 
-internal class ResumeRunnable(
-    private val continuation: Continuation<Unit>
+private class ResumeUndispatchedRunnable(
+    private val dispatcher: CoroutineDispatcher,
+    private val continuation: CancellableContinuation<Unit>
 ) : Runnable {
     override fun run() {
-        continuation.resume(Unit)
+        with(continuation) { dispatcher.resumeUndispatched(Unit) }
     }
 }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/ThreadPoolDispatcher.kt

@@ -46,33 +46,26 @@ fun newFixedThreadPoolContext(nThreads: Int, name: String, parent: Job? = null):
 }
 
 internal class PoolThread(
-    val dispatcher: ThreadPoolDispatcher, // for debugging & tests
+    @JvmField val dispatcher: ThreadPoolDispatcher, // for debugging & tests
     target: Runnable, name: String
 ) : Thread(target, name) {
     init { isDaemon = true }
 }
 
 internal class ThreadPoolDispatcher(
-        nThreads: Int,
-        name: String,
-        val job: Job
-) : CoroutineDispatcher(), Delay {
-    val threadNo = AtomicInteger()
-    val executor: ScheduledExecutorService = Executors.newScheduledThreadPool(nThreads) { target ->
+    private val nThreads: Int,
+    private val name: String,
+    job: Job
+) : ExecutorCoroutineDispatcherBase() {
+    private val threadNo = AtomicInteger()
+
+    override val executor: ScheduledExecutorService = Executors.newScheduledThreadPool(nThreads) { target ->
         PoolThread(this, target, if (nThreads == 1) name else name + "-" + threadNo.incrementAndGet())
     }
 
     init {
         job.invokeOnCompletion { executor.shutdown() }
     }
 
-    override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
-
-    override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
-        val timeout = executor.schedule(ResumeUndispatchedRunnable(this, continuation), time, unit)
-        continuation.cancelFutureOnCompletion(timeout)
-    }
-
-    override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
-        DisposableFutureHandle(executor.schedule(block, time, unit))
+    override fun toString(): String = "ThreadPoolDispatcher[$nThreads, $name]"
 }