我怎样才能将 throwable 传播到 Completable 的下一个链？

Question

我正在使用 RxJava 开发 Android 应用程序。

我有一些 API 调用链。

验证
消费

val verify = Completable.error(Exception("TEST"))
            .doOnSubscribe { Log.d(TAG, "1. verify") }
            .doOnComplete{ Log.d(TAG, "1. verify - success") }
            .doOnError { Log.e(TAG, "1. verify - failed: ${it.message}") }
            .retryWhen { attempts ->
                attempts.zipWith(
                    Flowable.range(1, 3), BiFunction<Throwable, Int, Long> { t, i ->
                        if (i <= 3) {
                            1L
                        } else {
                            throw t
                        }
                    }
                ).flatMap {
                    Flowable.timer(it, TimeUnit.SECONDS)
                }
            }

// 2. consume
val consume = Single.just("SUCCESS")
    .doOnSubscribe { Log.d(TAG, "2. consume") }
    .doOnSuccess { Log.d(TAG, "2. consume - success") }
    .doOnError { Log.e(TAG, "2. consume - failed: ${it.message}", it) }

disposable.add(
    verify.andThen (consume)
        .subscribeOn(ioScheduler)
        .observeOn(uiScheduler)
        .subscribe({
            Log.d(TAG, "done")
        }, { t ->
            Log.e(TAG, "failed: ${t.message}", t)
        })
);

我排除的是...

"verify" 应该每 1 秒调用 3 次。

3次重试失败，应该是Error。

但就我而言，"consume" 也是运行。

为什么？

如果 "verify" 失败，我想跳过 "consume"！

我该怎么做？

Answer 1

这是因为您的代码没有失败。

使用 Flowable.range(1, 3) 你创建了一个从 1 到 3 的范围，所以你的代码的 else 部分永远不会到达。

尝试使用 Flowable.range(1, 4)，您会看到正确的行为。

我怎样才能将 throwable 传播到 Completable 的下一个链？

How can I propagate throwable to the Completable's next chain?

android

rx-java

rx-java2