Complete mono { } on cancellation

[dkhalanskyjb]

Fixes #2262

[qwwdfsad]

Also, we've discussed that such behaviour creates unwanted inconsistency between rxMaybe and rxSingle even if the former always completes with a value

[qwwdfsad]

These functions seem to be misplaced and should belong to test source-set

[dkhalanskyjb]

There's a request for these functions (#1587), so I thought it would make sense to add them here (though this PR is unrelated) and to promote them to public later, after adding documentation and tests for them.

[qwwdfsad]

With such a check, we have a scenario when an exception is swallowed.
E.g. the following test:

@Test
fun testSwallowedException() = runTest {
    val started = CountDownLatch(1)
    val disposed = CountDownLatch(1)
    val m = mono {
        started.countDown()
        disposed.await()
        println("Running")
        throw ArithmeticException()
    }
    val d = m.subscribe()
    started.await()
    d.dispose()
    disposed.countDown()
}

will print Running and then will completely ignore the arithmetic exception

[qwwdfsad]

PTAL on discussion here #252 and the corresponding commit that fixes the problem

[dkhalanskyjb]

I have read the discussion and am not sure how it is applicable. Could you please clarify what behavior would you expect instead?

What happens here is that onError gets called after m is disposed of, and Reactor deals with errors from such objects by passing them to the onOperatorError hook, which is unset in this test. Setting the hook leads to the exception being reported:

    @Test
    fun testSwallowedException() = runTest {
        Hooks.onOperatorError("testSwallowedException") { t, a ->
            expectUnreached()
            t
        }
        val started = CountDownLatch(1)
        val disposed = CountDownLatch(1)
        val m = mono {
            started.countDown()
            disposed.await()
            println("Running")
            throw ArithmeticException()
        }
        val d = m.subscribe()
        started.await()
        d.dispose()
        disposed.countDown()
    }

successfully fails.

Moreover, it looks like using onOperatorError is the approach we take for undeliverable exceptions in Reactor, so… this exception is already dealt with the way we usually deal with the undeliverable exceptions. What am I missing here?

[qwwdfsad]

Oh, sorry, my bad, I've misinterpreted the code. Thanks for the clarification!