Introduce ExecutorRule for stress-tests for simpler development process

Author: qwwdfsad

kotlinx-coroutines-core/jvm/test/AwaitStressTest.kt

@@ -11,12 +11,8 @@ import java.util.concurrent.*
 class AwaitStressTest : TestBase() {
 
     private val iterations = 50_000 * stressTestMultiplier
-    private val pool = newFixedThreadPoolContext(4, "AwaitStressTest")
-
-    @After
-    fun tearDown() {
-        pool.close()
-    }
+    @get:Rule
+    public val pool =  ExecutorRule(4)
 
     @Test
     fun testMultipleExceptions() = runTest {

kotlinx-coroutines-core/jvm/test/CancellableContinuationResumeCloseStressTest.kt

@@ -11,20 +11,16 @@ import kotlin.test.*
 import kotlin.test.Test
 
 class CancellableContinuationResumeCloseStressTest : TestBase() {
-    private val dispatcher =
-        newFixedThreadPoolContext(2, "CancellableContinuationResumeCloseStressTest")
+    @get:Rule
+    public val dispatcher = ExecutorRule(2)
+
     private val startBarrier = CyclicBarrier(3)
     private val doneBarrier = CyclicBarrier(2)
     private val nRepeats = 1_000 * stressTestMultiplier
 
     private val closed = atomic(false)
     private var returnedOk = false
 
-    @After
-    fun tearDown() {
-        dispatcher.close()
-    }
-
     @Test
     @Suppress("BlockingMethodInNonBlockingContext")
     fun testStress() = runTest {
@@ -65,4 +61,4 @@ class CancellableContinuationResumeCloseStressTest : TestBase() {
     fun close() {
         assertFalse(closed.getAndSet(true))
     }
-}
+}

kotlinx-coroutines-core/jvm/test/ExecutorRule.kt

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import org.junit.rules.*
+import org.junit.runner.*
+import org.junit.runners.model.*
+import java.util.concurrent.*
+import kotlin.coroutines.*
+
+class ExecutorRule(private val numberOfThreads: Int) : TestRule, ExecutorCoroutineDispatcher() {
+
+    private var _executor: ExecutorCoroutineDispatcher? = null
+    override val executor: Executor
+        get() = _executor?.executor ?: error("Executor is not initialized")
+
+    override fun apply(base: Statement, description: Description): Statement {
+        return object : Statement() {
+            override fun evaluate() {
+                val threadPrefix = description.className.substringAfterLast(".") + "#" + description.methodName
+                _executor = newFixedThreadPoolContext(numberOfThreads, threadPrefix)
+                ignoreLostThreads(threadPrefix)
+                try {
+                    return base.evaluate()
+                } finally {
+                    val service = executor as ExecutorService
+                    service.shutdown()
+                    if (!service.awaitTermination(10, TimeUnit.SECONDS)) {
+                        error("Test $description timed out")
+                    }
+                }
+            }
+        }
+    }
+
+    override fun dispatch(context: CoroutineContext, block: Runnable) {
+        _executor?.dispatch(context, block) ?: error("Executor is not initialized")
+    }
+
+    override fun close() {
+        error("Cannot be closed manually")
+    }
+}