如何测试CompletableEmitter.tryOnError()？附相关实现代码

首先得先把你代码里的catch块补全，让tryOnError()真正发挥作用，同时别忘了执行传入的finally逻辑——毕竟你的函数设计里有这个参数：

inline fun completable(crossinline action: () -> Unit) = completable(action, {})
inline fun completable(
    crossinline action: () -> Unit,
    crossinline finally: () -> Unit
): Completable {
    return Completable.create { emitter ->
        try {
            action()
            emitter.onComplete()
        } catch (t: Throwable) {
            // 这里补上tryOnError的调用
            emitter.tryOnError(t)
        } finally {
            // 执行传入的finally逻辑
            finally()
        }
    }
}

接下来，我们用RxJava自带的TestObserver来测试tryOnError()的核心行为——它只会在**下游订阅者仍处于活跃状态（未取消、未完成、未发生错误）**时发送错误并返回true，否则返回false且不会发送错误。

场景1：下游活跃时，tryOnError成功发送错误 #

这个场景验证正常情况下，异常被正确传递到下游：

import io.reactivex.rxjava3.core.Completable
import io.reactivex.rxjava3.observers.TestObserver
import org.junit.Test
import kotlin.test.assert

class CompletableTest {

    @Test
    fun `when action throws exception and downstream is active, tryOnError sends error`() {
        val testException = RuntimeException("Expected test error")
        var finallyExecuted = false

        // 创建我们的Completable，action抛出异常，finally标记执行状态
        val testCompletable = completable(
            action = { throw testException },
            finally = { finallyExecuted = true }
        )

        // 用TestObserver订阅
        val testObserver = testCompletable.test()

        // 验证：收到了预期的异常，finally被执行
        testObserver.assertError(testException)
        assert(finallyExecuted)
    }
}

场景2：下游已取消时，tryOnError不会发送错误 #

这个场景要模拟下游已经取消订阅的情况，验证tryOnError()返回false且错误不会被传递：

@Test
fun `when downstream is cancelled, tryOnError returns false and does not send error`() {
    val testException = RuntimeException("Should not be sent")
    var tryOnErrorResult = false
    var finallyExecuted = false

    // 直接创建Completable模拟下游取消的场景
    val testCompletable = Completable.create { emitter ->
        // 先手动取消emitter，模拟下游已经取消订阅
        emitter.dispose()
        try {
            throw testException
        } catch (t: Throwable) {
            // 记录tryOnError的返回结果
            tryOnErrorResult = emitter.tryOnError(t)
        } finally {
            finallyExecuted = true
        }
    }

    val testObserver = testCompletable.test()

    // 验证：tryOnError返回false，下游没有收到错误，finally被执行
    assert(!tryOnErrorResult)
    testObserver.assertNoErrors()
    testObserver.assertNotComplete()
    assert(finallyExecuted)
}

补充：用TestObserver.cancel()模拟下游取消 #

如果你想更贴合自己的completable()函数测试下游取消场景，也可以通过TestObserver的cancel()方法提前取消订阅：

@Test
fun `when test observer cancels before exception, tryOnError does nothing`() {
    val testException = RuntimeException("Should not reach observer")
    var tryOnErrorResult = false
    var finallyExecuted = false

    val testCompletable = Completable.create { emitter ->
        try {
            // 等待一下，方便下游先取消订阅
            Thread.sleep(100)
            throw testException
        } catch (t: Throwable) {
            tryOnErrorResult = emitter.tryOnError(t)
        } finally {
            finallyExecuted = true
        }
    }

    val testObserver = testCompletable.test()
    // 立即取消订阅
    testObserver.cancel()

    // 等待异步操作完成
    Thread.sleep(200)

    assert(!tryOnErrorResult)
    testObserver.assertNoErrors()
    assert(finallyExecuted)
}

核心思路就是覆盖tryOnError()的两种核心分支：下游活跃和非活跃，用TestObserver来验证事件的发送情况，同时确保finally逻辑始终被执行。

内容的提问来源于stack exchange，提问作者Eugen Martynov