kotlin - 如何测试Completable Emitter.tryOnError()

Question

我有下一个代码:

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) {
            // Attempts to emit the specified {@code Throwable} error if the downstream
            // hasn't cancelled the sequence or is otherwise terminated, returning false
            // if the emission is not allowed to happen due to lifecycle restrictions.
            // <p>
            // Unlike {@link #onError(Throwable)}, the {@code RxJavaPlugins.onError} is not called
            // if the error could not be delivered.
            emitter.tryOnError(t)
        } finally {
            finally()
        }
    }
}

我想编写一个测试，证明错误没有传递给ErrorHandler。所以我尝试了类似的东西:

@Test
fun `Completable do not crash when terminated`() {
    var ex: Throwable? = null
    RxJavaPlugins.setErrorHandler { t -> ex = t }

    Completable.mergeArray(
            completable {
                throw IOException("test1")
            }.subscribeOn(Schedulers.io()),
            completable {
                throw IOException("test2")
            }.subscribeOn(Schedulers.io())
    ).onErrorComplete()
            .blockingAwait()

    RxJavaPlugins.setErrorHandler(null)

    ex?.let {
        throw it
    }
}

如果将其传递给处理程序，则应引发异常。

不幸的是，该测试不稳定。我认为在这种情况下，多线程确实很棘手。

我可以改善这项测试吗？也许用 TestScheduler精确模拟了第一个异常终止流之后第二个异常着陆的情况？

Answer 1

您不需要为此进行并发，但是在抛出异常之前先让使用者处理掉:

List<Throwable> errors = new ArrayList<>();

RxJavaPlugins.setErrorHandler(errors::add);

try {
    TestObserver<Void> to = new TestObserver<>();

    completable(() -> {
        to.dispose();
        throw new RuntimeException();
    })
    .subscribe(to);
} finally {
    RxJavaPlugins.setErrorHandler(null);
}

assertTrue(errors.isEmpty());