Add some tests and fixes

* Fixed `doLockedNext` not releasing the lock in
  `PublisherCoroutine` if `null` is emitted
* Fixed `flux`, `publish`, `rxObservable` and `rxFlowable`
  incorrectly reporting `isClosedForSend == true` just after
  closing the channel

Author: dkhalanskyjb

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

@@ -5,6 +5,7 @@
 package kotlinx.coroutines.jdk9
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.exceptions.*
 import org.junit.Test
 import kotlinx.coroutines.flow.flowOn
 import org.junit.runner.*
@@ -132,21 +133,13 @@ class IntegrationTest(
 }
 
 internal suspend inline fun <reified E: Throwable> assertCallsExceptionHandlerWith(
-    crossinline operation: suspend (CoroutineExceptionHandler) -> Unit): E
-{
-    val caughtExceptions = mutableListOf<Throwable>()
-    val exceptionHandler = object: AbstractCoroutineContextElement(CoroutineExceptionHandler),
-        CoroutineExceptionHandler
-    {
-        override fun handleException(context: CoroutineContext, exception: Throwable) {
-            caughtExceptions += exception
-        }
-    }
-    return withContext(exceptionHandler) {
-        operation(exceptionHandler)
-        caughtExceptions.single().let {
-            assertTrue(it is E, it.toString())
-            it
+    crossinline operation: suspend (CoroutineExceptionHandler) -> Unit): E =
+    CapturingHandler().let { handler ->
+        withContext(handler) {
+            operation(handler)
+            handler.getException().let {
+                assertTrue(it is E, it.toString())
+                it
+            }
         }
-    }
-}
+    }

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

@@ -5,7 +5,11 @@
 package kotlinx.coroutines.jdk9
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.*
 import org.junit.Test
+import java.lang.NullPointerException
+import java.util.concurrent.*
+import java.util.concurrent.CancellationException
 import java.util.concurrent.Flow as JFlow
 import kotlin.test.*
 
@@ -121,6 +125,25 @@ class PublishTest : TestBase() {
         finish(7)
     }
 
+    /** Tests that, as soon as `ProducerScope.close` is called, `isClosedForSend` starts returning `true`. */
+    @Test
+    fun testChannelClosing() = runTest {
+        expect(1)
+        val publisher = flowPublish<Int>(Dispatchers.Unconfined) {
+            expect(3)
+            close()
+            assert(isClosedForSend)
+            expect(4)
+        }
+        try {
+            expect(2)
+            publisher.awaitFirstOrNull()
+        } catch (e: CancellationException) {
+            expect(5)
+        }
+        finish(6)
+    }
+
     @Test
     fun testOnNextError() = runTest {
         val latch = CompletableDeferred<Unit>()
@@ -130,9 +153,10 @@ class PublishTest : TestBase() {
                 expect(4)
                 try {
                     send("OK")
-                } catch(e: Throwable) {
+                } catch (e: Throwable) {
                     expect(6)
                     assert(e is TestException)
+                    assert(isClosedForSend)
                     latch.complete(Unit)
                 }
             }
@@ -162,6 +186,51 @@ class PublishTest : TestBase() {
         finish(7)
     }
 
+    /** Tests the behavior when a call to `onNext` fails after the channel is already closed. */
+    @Test
+    fun testOnNextErrorAfterCancellation() = runTest {
+        assertCallsExceptionHandlerWith<TestException> { handler ->
+            var producerScope: ProducerScope<Int>? = null
+            CompletableDeferred<Unit>()
+            expect(1)
+            var job: Job? = null
+            val publisher = flowPublish<Int>(handler + Dispatchers.Unconfined) {
+                producerScope = this
+                expect(4)
+                job = launch {
+                    delay(Long.MAX_VALUE)
+                }
+            }
+            expect(2)
+            publisher.subscribe(object: JFlow.Subscriber<Int> {
+                override fun onSubscribe(s: JFlow.Subscription) {
+                    expect(3)
+                    s.request(Long.MAX_VALUE)
+                }
+                override fun onNext(t: Int) {
+                    expect(6)
+                    assertEquals(1, t)
+                    job!!.cancel()
+                    throw TestException()
+                }
+                override fun onError(t: Throwable?) {
+                    /* Correct changes to the implementation could lead to us entering or not entering this method, but
+                    it only matters that if we do, it is the "correct" exception that was validly used to cancel the
+                    coroutine that gets passed here and not `TestException`. */
+                    assertTrue(t is CancellationException)
+                }
+                override fun onComplete() { expectUnreached() }
+            })
+            expect(5)
+            val result: ChannelResult<Unit> = producerScope!!.trySend(1)
+            val e = result.exceptionOrNull()!!
+            assertTrue(e is CancellationException, "The actual error: $e")
+            assertTrue(producerScope!!.isClosedForSend)
+            assertTrue(result.isFailure)
+        }
+        finish(7)
+    }
+
     @Test
     fun testFailingConsumer() = runTest {
         val pub = flowPublish(currentDispatcher()) {
@@ -183,4 +252,39 @@ class PublishTest : TestBase() {
     fun testIllegalArgumentException() {
         assertFailsWith<IllegalArgumentException> { flowPublish<Int>(Job()) { } }
     }
+
+    /** Tests that `trySend` doesn't throw in `flowPublish`. */
+    @Test
+    fun testTrySendNotThrowing() = runTest {
+        var producerScope: ProducerScope<Int>? = null
+        expect(1)
+        val publisher = flowPublish<Int>(Dispatchers.Unconfined) {
+            producerScope = this
+            expect(3)
+            delay(Long.MAX_VALUE)
+        }
+        val job = launch(start = CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            publisher.awaitFirstOrNull()
+            expectUnreached()
+        }
+        job.cancel()
+        expect(4)
+        val result = producerScope!!.trySend(1)
+        assertTrue(result.isFailure)
+        finish(5)
+    }
+
+    /** Tests that all methods on `flowPublish` fail without closing the channel when attempting to emit `null`. */
+    @Test
+    fun testEmittingNull() = runTest {
+        val publisher = flowPublish {
+            assertFailsWith<NullPointerException> { send(null) }
+            assertFailsWith<NullPointerException> { trySend(null) }
+            @Suppress("DEPRECATION")
+            assertFailsWith<NullPointerException> { offer(null) }
+            send("OK")
+        }
+        assertEquals("OK", publisher.awaitFirstOrNull())
+    }
 }

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

@@ -75,7 +75,7 @@ public class PublisherCoroutine<in T>(
     @Volatile
     private var cancelled = false // true after Subscription.cancel() is invoked
 
-    override val isClosedForSend: Boolean get() = isCompleted
+    override val isClosedForSend: Boolean get() = !isActive
     override fun close(cause: Throwable?): Boolean = cancelCoroutine(cause)
     override fun invokeOnClose(handler: (Throwable?) -> Unit): Nothing =
         throw UnsupportedOperationException("PublisherCoroutine doesn't support invokeOnClose")
@@ -133,6 +133,7 @@ public class PublisherCoroutine<in T>(
      */
     private fun doLockedNext(elem: T): Throwable? {
         if (elem == null) {
+            unlockAndCheckCompleted()
             throw NullPointerException("Attempted to emit `null` inside a reactive publisher")
         }
         /** This guards against the case when the caller of this function managed to lock the mutex not because some
@@ -212,7 +213,7 @@ public class PublisherCoroutine<in T>(
             _nRequested.value = SIGNALLED // we'll signal onError/onCompleted (the final state, so no CAS needed)
             // Specification requires that after the cancellation is requested we eventually stop calling onXXX
             if (cancelled) {
-                // If the parent had failed to handle our exception, then we must not lose this exception
+                // If the parent failed to handle this exception, then we must not lose the exception
                 if (cause != null && !handled) exceptionOnCancelHandler(cause, context)
                 return
             }

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

@@ -208,7 +208,11 @@ public class FlowSubscription<T>(
             consumeFlow()
             true
         } catch (cause: Throwable) {
+            /* TODO: The part after "||" is a hack needed due to [cause] having travelled over a coroutine boundary and
+                being changed from the result of [getCancellationException()]. */
             if (cancellationRequested && !isActive && (cause === getCancellationException() || cause.cause === getCancellationException() && cause.message == getCancellationException().message)) {
+                /* TODO: This is incorrect, as [Subscriber.onComplete] denotes the end of the stream and not just any
+                    non-erroneous terminal state. */
                 subscriber.onComplete()
             } else {
                 try {

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

@@ -5,12 +5,14 @@
 package kotlinx.coroutines.reactive
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.exceptions.*
 import org.junit.Test
 import org.junit.runner.*
 import org.junit.runners.*
 import org.reactivestreams.*
 import java.lang.IllegalStateException
 import java.lang.RuntimeException
+import kotlin.contracts.*
 import kotlin.coroutines.*
 import kotlin.test.*
 
@@ -235,20 +237,16 @@ class IntegrationTest(
 
 }
 
+@OptIn(ExperimentalContracts::class)
 internal suspend inline fun <reified E: Throwable> assertCallsExceptionHandlerWith(
-    crossinline operation: suspend (CoroutineExceptionHandler) -> Unit): E
-{
-    val caughtExceptions = mutableListOf<Throwable>()
-    val exceptionHandler = object: AbstractCoroutineContextElement(CoroutineExceptionHandler),
-        CoroutineExceptionHandler
-    {
-        override fun handleException(context: CoroutineContext, exception: Throwable) {
-            caughtExceptions += exception
-        }
+    crossinline operation: suspend (CoroutineExceptionHandler) -> Unit): E {
+    contract {
+        callsInPlace(operation, InvocationKind.EXACTLY_ONCE)
     }
-    return withContext(exceptionHandler) {
-        operation(exceptionHandler)
-        caughtExceptions.single().let {
+    val handler = CapturingHandler()
+    return withContext(handler) {
+        operation(handler)
+        handler.getException().let {
             assertTrue(it is E, it.toString())
             it
         }