Actually test 'onLock' and the corresponding concurrency and cancellation of Reactive's onSend

   * Also, cleanup the related code a bit

Author: qwwdfsad

kotlinx-coroutines-core/common/src/channels/Channel.kt

@@ -64,7 +64,6 @@ public interface SendChannel<in E> {
      */
     public val onSend: SelectClause2<E, SendChannel<E>>
 
-
     /**
      * Immediately adds the specified [element] to this channel, if this doesn't violate its capacity restrictions,
      * and returns the successful result. Otherwise, returns failed or closed result.

kotlinx-coroutines-core/jvm/test/channels/RandevouzChannelStressTest.kt renamed to kotlinx-coroutines-core/jvm/test/channels/RendezvousChannelStressTest.kt

@@ -7,7 +7,7 @@ package kotlinx.coroutines.channels
 import kotlinx.coroutines.*
 import org.junit.*
 
-class RandevouzChannelStressTest : TestBase() {
+class RendezvousChannelStressTest : TestBase() {
 
     @Test
     fun testStress() = runTest {

reactive/kotlinx-coroutines-reactive/test/PublishTest.kt

@@ -5,9 +5,12 @@
 package kotlinx.coroutines.reactive
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.flow.*
 import org.junit.Test
 import org.reactivestreams.*
+import java.util.concurrent.*
 import kotlin.test.*
 
 class PublishTest : TestBase() {
@@ -284,4 +287,37 @@ class PublishTest : TestBase() {
         }
         assertEquals("OK", publisher.awaitFirstOrNull())
     }
-}
+
+    @Test
+    fun testOnSendCancelled() = runTest {
+        val latch = CountDownLatch(1)
+        val published = publish(Dispatchers.Default) {
+            expect(2)
+            // Collector is ready
+            send(1)
+            expect(3)
+            try {
+                send(2)
+                expectUnreached()
+            } catch (e: CancellationException) {
+                expect(7)
+                // publisher cancellation is async
+                latch.countDown()
+                throw e
+            }
+        }
+
+        expect(1)
+        val job = launch {
+            published.asFlow().buffer(0).collect {
+                expect(4)
+                hang { expect(6) }
+            }
+        }
+        yield()
+        expect(5)
+        job.cancelAndJoin()
+        latch.await()
+        finish(8)
+    }
+}

reactive/kotlinx-coroutines-reactive/test/PublisherMultiTest.kt

@@ -5,6 +5,7 @@
 package kotlinx.coroutines.reactive
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.selects.*
 import org.junit.Test
 import kotlin.test.*
 
@@ -16,7 +17,7 @@ class PublisherMultiTest : TestBase() {
             // concurrent emitters (many coroutines)
             val jobs = List(n) {
                 // launch
-                launch {
+                launch(Dispatchers.Default) {
                     send(it)
                 }
             }
@@ -28,4 +29,26 @@ class PublisherMultiTest : TestBase() {
         }
         assertEquals(n, resultSet.size)
     }
+
+    @Test
+    fun testConcurrentStressOnSend() = runBlocking {
+        val n = 10_000 * stressTestMultiplier
+        val observable = publish {
+            // concurrent emitters (many coroutines)
+            val jobs = List(n) {
+                // launch
+                launch(Dispatchers.Default) {
+                    select<Unit> {
+                        onSend(it) {}
+                    }
+                }
+            }
+            jobs.forEach { it.join() }
+        }
+        val resultSet = mutableSetOf<Int>()
+        observable.collect {
+            assertTrue(resultSet.add(it))
+        }
+        assertEquals(n, resultSet.size)
+    }
 }

reactive/kotlinx-coroutines-rx2/test/ObservableCompletionStressTest.kt

@@ -12,7 +12,7 @@ import kotlin.coroutines.*
 class ObservableCompletionStressTest : TestBase() {
     private val N_REPEATS = 10_000 * stressTestMultiplier
 
-    private fun CoroutineScope.range(context: CoroutineContext, start: Int, count: Int) = rxObservable(context) {
+    private fun range(context: CoroutineContext, start: Int, count: Int) = rxObservable(context) {
         for (x in start until start + count) send(x)
     }
 
@@ -33,4 +33,4 @@ class ObservableCompletionStressTest : TestBase() {
             }
         }
     }
-}
+}

reactive/kotlinx-coroutines-rx2/test/ObservableMultiTest.kt

@@ -6,6 +6,7 @@ package kotlinx.coroutines.rx2
 
 import io.reactivex.*
 import kotlinx.coroutines.*
+import kotlinx.coroutines.selects.*
 import org.junit.Test
 import java.io.*
 import kotlin.test.*
@@ -47,6 +48,29 @@ class ObservableMultiTest : TestBase() {
         }
     }
 
+    @Test
+    fun testConcurrentStressOnSend() {
+        val n = 10_000 * stressTestMultiplier
+        val observable = rxObservable {
+            newCoroutineContext(coroutineContext)
+            // concurrent emitters (many coroutines)
+            val jobs = List(n) {
+                // launch
+                launch(Dispatchers.Default) {
+                    val i = it
+                    select<Unit> {
+                        onSend(i) {}
+                    }
+                }
+            }
+            jobs.forEach { it.join() }
+        }
+        checkSingleValue(observable.toList()) { list ->
+            assertEquals(n, list.size)
+            assertEquals((0 until n).toList(), list.sorted())
+        }
+    }
+
     @Test
     fun testIteratorResendUnconfined() {
         val n = 10_000 * stressTestMultiplier
@@ -88,4 +112,4 @@ class ObservableMultiTest : TestBase() {
             assertEquals("OK", it)
         }
     }
-}
+}

reactive/kotlinx-coroutines-rx3/test/ObservableMultiTest.kt

@@ -6,6 +6,7 @@ package kotlinx.coroutines.rx3
 
 import io.reactivex.rxjava3.core.*
 import kotlinx.coroutines.*
+import kotlinx.coroutines.selects.*
 import org.junit.Test
 import java.io.*
 import kotlin.test.*
@@ -34,7 +35,7 @@ class ObservableMultiTest : TestBase() {
             // concurrent emitters (many coroutines)
             val jobs = List(n) {
                 // launch
-                launch {
+                launch(Dispatchers.Default) {
                     val i = it
                     send(i)
                 }
@@ -47,6 +48,29 @@ class ObservableMultiTest : TestBase() {
         }
     }
 
+    @Test
+    fun testConcurrentStressOnSend() {
+        val n = 10_000 * stressTestMultiplier
+        val observable = rxObservable {
+            newCoroutineContext(coroutineContext)
+            // concurrent emitters (many coroutines)
+            val jobs = List(n) {
+                // launch
+                launch(Dispatchers.Default) {
+                    val i = it
+                    select<Unit> {
+                        onSend(i) {}
+                    }
+                }
+            }
+            jobs.forEach { it.join() }
+        }
+        checkSingleValue(observable.toList()) { list ->
+            assertEquals(n, list.size)
+            assertEquals((0 until n).toList(), list.sorted())
+        }
+    }
+
     @Test
     fun testIteratorResendUnconfined() {
         val n = 10_000 * stressTestMultiplier