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

/*
 * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */

package kotlinx.coroutines.rx2

import io.reactivex.*
import io.reactivex.exceptions.*
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.reactive.*
import org.junit.Test
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import kotlin.test.*

// Check that exception is not leaked to the global exception handler
class LeakedExceptionTest : TestBase() {

    private val handler: (Throwable) -> Unit =
        { assertTrue { it is UndeliverableException && it.cause is TestException } }

    @Test
    fun testSingle() = withExceptionHandler(handler) {
        withFixedThreadPool(4) { dispatcher ->
            val flow = rxSingle<Unit>(dispatcher) { throw TestException() }.toFlowable().asFlow()
            runBlocking {
                repeat(10000) {
                    combine(flow, flow) { _, _ -> Unit }
                        .catch {}
                        .collect {}
                }
            }
        }
    }

    @Test
    fun testObservable() = withExceptionHandler(handler) {
        withFixedThreadPool(4) { dispatcher ->
            val flow = rxObservable<Unit>(dispatcher) { throw TestException() }
                .toFlowable(BackpressureStrategy.BUFFER)
                .asFlow()
            runBlocking {
                repeat(10000) {
                    combine(flow, flow) { _, _ -> Unit }
                        .catch {}
                        .collect {}
                }
            }
        }
    }

    @Test
    fun testFlowable() = withExceptionHandler(handler) {
        withFixedThreadPool(4) { dispatcher ->
            val flow = rxFlowable<Unit>(dispatcher) { throw TestException() }.asFlow()
            runBlocking {
                repeat(10000) {
                    combine(flow, flow) { _, _ -> Unit }
                        .catch {}
                        .collect {}
                }
            }
        }
    }

    /**
     * This test doesn't test much and was added to display a problem with straighforward use of
     * [withExceptionHandler].
     *
     * If one was to remove `dispatcher` and launch `rxFlowable` with an empty coroutine context,
     * this test would fail fairly often, while other tests were also vulnerable, but the problem is
     * much more difficult to reproduce. Thus, this test is a justification for adding `dispatcher`
     * to other tests.
     *
     * See the commit that introduced this test for a better explanation.
     */
    @Test
    fun testResettingExceptionHandler() = withExceptionHandler(handler) {
        withFixedThreadPool(4) { dispatcher ->
            val flow = rxFlowable<Unit>(dispatcher) {
                if ((0..1).random() == 0) {
                    Thread.sleep(100)
                }
                throw TestException()
            }.asFlow()
            runBlocking {
                combine(flow, flow) { _, _ -> Unit }
                    .catch {}
                    .collect {}
            }
        }
    }

    /**
     * Run in a thread pool, then wait for all the tasks to finish.
     */
    private fun withFixedThreadPool(numberOfThreads: Int, block: (CoroutineDispatcher) -> Unit) {
        val pool = Executors.newFixedThreadPool(numberOfThreads)
        val dispatcher = pool.asCoroutineDispatcher()
        block(dispatcher)
        pool.shutdown()
        while (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
            /* deliberately empty */
        }
    }
}