Implement some tests for the new functionality

Author: dkhalanskyjb

integration/kotlinx-coroutines-guava/test/ListenableFutureTest.kt

@@ -9,6 +9,7 @@ import org.junit.Test
 import java.util.concurrent.*
 import java.util.concurrent.CancellationException
 import java.util.concurrent.atomic.*
+import kotlinx.coroutines.testing.CountDownLatch
 import kotlin.test.*
 
 class ListenableFutureTest : TestBase() {

kotlinx-coroutines-core/concurrent/test/DefaultDelayTest.kt

@@ -0,0 +1,17 @@
+package kotlinx.coroutines
+
+import kotlin.test.*
+import kotlinx.coroutines.testing.*
+
+class DefaultDelayTest: TestBase() {
+    @Test
+    fun testDelayOnUnconfined() = runTest {
+        val latch = CountDownLatch(1)
+        launch(Dispatchers.Unconfined) {
+            delay(1)
+            latch.await()
+        }
+        delay(10)
+        latch.countDown()
+    }
+}

kotlinx-coroutines-core/concurrent/test/RunBlockingTest.kt

@@ -194,4 +194,22 @@ class RunBlockingTest : TestBase() {
             }
         }
     }
+
+    /** Tests that tasks scheduled on a closed `runBlocking` event loop get processed in an I/O thread. */
+    @OptIn(ExperimentalStdlibApi::class)
+    @Test
+    fun testLeakedEventLoopGetsProcessedInIO() {
+        val dispatcher = runBlocking {
+            coroutineContext[CoroutineDispatcher.Key]
+        }!!
+        runBlocking {
+            GlobalScope.launch(dispatcher) {
+                assertTrue(runningOnIoThread())
+                delay(1.milliseconds)
+                assertTrue(runningOnIoThread())
+            }.join()
+        }
+    }
 }
+
+internal expect fun runningOnIoThread(): Boolean

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

@@ -2,9 +2,10 @@ package kotlinx.coroutines
 
 import kotlinx.coroutines.testing.*
 import org.junit.Test
-import java.util.concurrent.*
 import kotlin.coroutines.*
 import kotlin.test.*
+import java.util.concurrent.Executors
+import java.util.concurrent.RejectedExecutionException
 
 class ExecutorsTest : TestBase() {
     private fun checkThreadName(prefix: String) {
@@ -45,7 +46,7 @@ class ExecutorsTest : TestBase() {
 
     @Test
     fun testConvertedDispatcherToExecutor() {
-        val executor: ExecutorService = Executors.newSingleThreadExecutor { r -> Thread(r, "TestExecutor") }
+        val executor = Executors.newSingleThreadExecutor { r -> Thread(r, "TestExecutor") }
         val dispatcher: CoroutineDispatcher = executor.asCoroutineDispatcher()
         assertSame(executor, dispatcher.asExecutor())
         executor.shutdown()

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

@@ -5,9 +5,10 @@ import org.junit.*
 import org.junit.Test
 import org.junit.runner.*
 import org.junit.runners.*
-import java.util.concurrent.*
+import java.util.concurrent.Executors
 import kotlin.coroutines.*
 import kotlin.test.*
+import kotlin.time.Duration.Companion.seconds
 
 @RunWith(Parameterized::class)
 class FailingCoroutinesMachineryTest(
@@ -139,7 +140,7 @@ class FailingCoroutinesMachineryTest(
     }
 
     private fun checkException() {
-        latch.await(2, TimeUnit.SECONDS)
+        latch.await(2.seconds)
         val e = caught
         assertNotNull(e)
         // First condition -- failure in context element

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

@@ -4,6 +4,7 @@ import kotlinx.coroutines.testing.*
 import java.util.concurrent.*
 import java.util.concurrent.atomic.*
 import kotlin.test.*
+import kotlinx.coroutines.testing.CountDownLatch
 
 /**
  * Testing the procedure of attaching a child to the parent job.

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

@@ -2,7 +2,6 @@ package kotlinx.coroutines
 
 import kotlinx.coroutines.testing.*
 import org.junit.Test
-import java.util.concurrent.CountDownLatch
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.*
 

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

@@ -2,6 +2,9 @@ package kotlinx.coroutines
 
 import kotlinx.coroutines.testing.*
 import org.junit.*
+import kotlin.test.Test
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.milliseconds
 
 class RunBlockingJvmTest : TestBase() {
     @Test
@@ -13,3 +16,5 @@ class RunBlockingJvmTest : TestBase() {
         rb.hashCode() // unused
     }
 }
+
+internal actual fun runningOnIoThread(): Boolean = Thread.currentThread().isIoDispatcherThread()

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

@@ -2,7 +2,6 @@ package kotlinx.coroutines
 
 import kotlinx.coroutines.testing.*
 import kotlinx.coroutines.sync.*
-import java.util.concurrent.*
 import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.*
 import kotlin.test.*
@@ -139,7 +138,7 @@ class ThreadLocalStressTest : TestBase() {
                     cancel()
                     semaphore.acquire()
                 }
-            } catch (e: CancellationException) {
+            } catch (_: CancellationException) {
                 // Ignore cancellation
             }
         }
@@ -154,7 +153,7 @@ class ThreadLocalStressTest : TestBase() {
                     cancel()
                     semaphore.acquire()
                 }
-            } catch (e: CancellationException) {
+            } catch (_: CancellationException) {
                 // Ignore cancellation
             }
         }

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

@@ -4,6 +4,7 @@ import kotlinx.coroutines.testing.*
 import org.junit.*
 import org.junit.Test
 import java.util.concurrent.*
+import kotlinx.coroutines.testing.CountDownLatch
 import kotlin.test.*
 
 class UnconfinedConcurrentStressTest : TestBase() {

kotlinx-coroutines-core/jvm/test/channels/InvokeOnCloseStressTest.kt

@@ -4,7 +4,6 @@ import kotlinx.coroutines.testing.*
 import kotlinx.coroutines.*
 import org.junit.*
 import org.junit.Test
-import java.util.concurrent.*
 import java.util.concurrent.atomic.*
 import kotlin.coroutines.*
 import kotlin.test.*

kotlinx-coroutines-core/jvm/test/jdk8/future/FutureTest.kt

@@ -14,6 +14,7 @@ import kotlin.concurrent.withLock
 import kotlin.coroutines.*
 import kotlin.reflect.*
 import kotlin.test.*
+import kotlinx.coroutines.testing.CountDownLatch
 
 class FutureTest : TestBase() {
     @Before

kotlinx-coroutines-core/jvm/test/scheduling/CoroutineSchedulerOversubscriptionTest.kt

@@ -2,9 +2,10 @@ package kotlinx.coroutines.scheduling
 
 import kotlinx.coroutines.testing.*
 import kotlinx.coroutines.*
-import org.junit.Test
 import java.util.concurrent.*
 import java.util.concurrent.atomic.AtomicInteger
+import kotlinx.coroutines.testing.CountDownLatch
+import kotlin.test.Test
 
 class CoroutineSchedulerOversubscriptionTest : TestBase() {
 

kotlinx-coroutines-core/jvm/test/scheduling/CoroutineSchedulerStressTest.kt

@@ -4,9 +4,7 @@ import kotlinx.coroutines.testing.*
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.internal.*
-import org.junit.*
-import org.junit.Test
-import java.util.concurrent.*
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.*
 import kotlin.coroutines.*
 import kotlin.test.*
@@ -27,7 +25,7 @@ class CoroutineSchedulerStressTest : TestBase() {
     private val processed = AtomicInteger(0)
     private val finishLatch = CountDownLatch(1)
 
-    @After
+    @AfterTest
     fun tearDown() {
         dispatcher.close()
     }

kotlinx-coroutines-core/jvm/test/scheduling/CoroutineSchedulerTest.kt

@@ -1,9 +1,6 @@
 package kotlinx.coroutines.scheduling
 
 import kotlinx.coroutines.testing.*
-import org.junit.Test
-import java.lang.Runnable
-import java.util.concurrent.*
 import kotlin.coroutines.*
 import kotlin.test.*
 

kotlinx-coroutines-core/jvm/test/scheduling/DefaultDispatchersTest.kt

@@ -5,6 +5,7 @@ import kotlinx.coroutines.*
 import org.junit.Test
 import java.util.concurrent.*
 import java.util.concurrent.atomic.*
+import kotlinx.coroutines.testing.CountDownLatch
 import kotlin.test.*
 
 class DefaultDispatchersTest : TestBase() {

kotlinx-coroutines-core/jvm/test/scheduling/WorkQueueStressTest.kt

@@ -1,10 +1,8 @@
 package kotlinx.coroutines.scheduling
 
 import kotlinx.coroutines.testing.*
-import kotlinx.coroutines.*
 import org.junit.*
 import org.junit.Test
-import java.util.concurrent.*
 import kotlin.concurrent.*
 import kotlin.jvm.internal.*
 import kotlin.test.*

kotlinx-coroutines-core/native/src/Dispatchers.kt

@@ -1,6 +1,7 @@
 package kotlinx.coroutines
 
 import kotlin.coroutines.*
+import kotlin.native.concurrent.Worker
 
 
 public actual object Dispatchers {

kotlinx-coroutines-core/native/test/RunBlockingNativeTest.kt

@@ -0,0 +1,7 @@
+package kotlinx.coroutines
+
+import kotlin.native.concurrent.ObsoleteWorkersApi
+import kotlin.native.concurrent.Worker
+
+@OptIn(ObsoleteWorkersApi::class)
+internal actual fun runningOnIoThread(): Boolean = Worker.current.name.startsWith("Dispatchers.IO")

kotlinx-coroutines-debug/test/RunningThreadStackMergeTest.kt

@@ -6,6 +6,7 @@ import kotlinx.coroutines.debug.internal.*
 import org.junit.Test
 import java.util.concurrent.*
 import kotlin.test.*
+import kotlinx.coroutines.testing.CountDownLatch
 
 class RunningThreadStackMergeTest : DebugTestBase() {
 

reactive/kotlinx-coroutines-jdk9/src/Await.kt

@@ -15,7 +15,7 @@ import kotlinx.coroutines.reactive.*
  *
  * @throws NoSuchElementException if the publisher does not emit any value
  */
-public suspend fun <T> Flow.Publisher<T>.awaitFirst(): T =
+suspend fun <T> Flow.Publisher<T>.awaitFirst(): T =
     FlowAdapters.toPublisher(this).awaitFirst()
 
 /**
@@ -27,7 +27,7 @@ public suspend fun <T> Flow.Publisher<T>.awaitFirst(): T =
  * If the [Job] of the current coroutine is cancelled while the suspending function is waiting, this
  * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  */
-public suspend fun <T> Flow.Publisher<T>.awaitFirstOrDefault(default: T): T =
+suspend fun <T> Flow.Publisher<T>.awaitFirstOrDefault(default: T): T =
     FlowAdapters.toPublisher(this).awaitFirstOrDefault(default)
 
 /**
@@ -38,7 +38,7 @@ public suspend fun <T> Flow.Publisher<T>.awaitFirstOrDefault(default: T): T =
  * If the [Job] of the current coroutine is cancelled while the suspending function is waiting, this
  * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  */
-public suspend fun <T> Flow.Publisher<T>.awaitFirstOrNull(): T? =
+suspend fun <T> Flow.Publisher<T>.awaitFirstOrNull(): T? =
     FlowAdapters.toPublisher(this).awaitFirstOrNull()
 
 /**
@@ -50,7 +50,7 @@ public suspend fun <T> Flow.Publisher<T>.awaitFirstOrNull(): T? =
  * If the [Job] of the current coroutine is cancelled while the suspending function is waiting, this
  * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
  */
-public suspend fun <T> Flow.Publisher<T>.awaitFirstOrElse(defaultValue: () -> T): T =
+suspend fun <T> Flow.Publisher<T>.awaitFirstOrElse(defaultValue: () -> T): T =
     FlowAdapters.toPublisher(this).awaitFirstOrElse(defaultValue)
 
 /**
@@ -63,7 +63,7 @@ public suspend fun <T> Flow.Publisher<T>.awaitFirstOrElse(defaultValue: () -> T)
  *
  * @throws NoSuchElementException if the publisher does not emit any value
  */
-public suspend fun <T> Flow.Publisher<T>.awaitLast(): T =
+suspend fun <T> Flow.Publisher<T>.awaitLast(): T =
     FlowAdapters.toPublisher(this).awaitLast()
 
 /**
@@ -77,5 +77,5 @@ public suspend fun <T> Flow.Publisher<T>.awaitLast(): T =
  * @throws NoSuchElementException if the publisher does not emit any value
  * @throws IllegalArgumentException if the publisher emits more than one value
  */
-public suspend fun <T> Flow.Publisher<T>.awaitSingle(): T =
+suspend fun <T> Flow.Publisher<T>.awaitSingle(): T =
     FlowAdapters.toPublisher(this).awaitSingle()

reactive/kotlinx-coroutines-jdk9/src/Publish.kt

@@ -28,7 +28,7 @@ import org.reactivestreams.FlowAdapters
  *
  * @throws IllegalArgumentException if the provided [context] contains a [Job] instance.
  */
-public fun <T> flowPublish(
+fun <T> flowPublish(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit
 ): Flow.Publisher<T> = FlowAdapters.toFlowPublisher(publish(context, block))

reactive/kotlinx-coroutines-jdk9/src/ReactiveFlow.kt

@@ -19,7 +19,7 @@ import java.util.concurrent.Flow as JFlow
  * If any of the resulting flow transformations fails, the subscription is immediately cancelled and all the in-flight
  * elements are discarded.
  */
-public fun <T : Any> JFlow.Publisher<T>.asFlow(): Flow<T> =
+fun <T : Any> JFlow.Publisher<T>.asFlow(): Flow<T> =
     FlowAdapters.toPublisher(this).asFlow()
 
 /**
@@ -32,7 +32,7 @@ public fun <T : Any> JFlow.Publisher<T>.asFlow(): Flow<T> =
  * is used, so calls are performed from an arbitrary thread.
  */
 @JvmOverloads // binary compatibility
-public fun <T : Any> Flow<T>.asPublisher(context: CoroutineContext = EmptyCoroutineContext): JFlow.Publisher<T> =
+fun <T : Any> Flow<T>.asPublisher(context: CoroutineContext = EmptyCoroutineContext): JFlow.Publisher<T> =
     FlowAdapters.toFlowPublisher(asReactivePublisher(context))
 
 /**
@@ -41,5 +41,5 @@ public fun <T : Any> Flow<T>.asPublisher(context: CoroutineContext = EmptyCorout
  * If [action] throws an exception at some point, the subscription is cancelled, and the exception is rethrown from
  * [collect]. Also, if the publisher signals an error, that error is rethrown from [collect].
  */
-public suspend inline fun <T> JFlow.Publisher<T>.collect(action: (T) -> Unit): Unit =
+suspend inline fun <T> JFlow.Publisher<T>.collect(action: (T) -> Unit): Unit =
     FlowAdapters.toPublisher(this).collect(action)

reactive/kotlinx-coroutines-jdk9/test/FlowAsPublisherTest.kt

@@ -82,7 +82,7 @@ class FlowAsPublisherTest : TestBase() {
         try {
             expect(1)
             publisher.awaitFirstOrNull()
-        } catch (e: CancellationException) {
+        } catch (_: CancellationException) {
             expect(3)
         }
         finish(4)