ReactiveX 系列文章目錄

onErrorResumeNext

在源 Observable 遇到錯誤時，立即停止源 Observable 的數(shù)據(jù)發(fā)送，并用新的 Observable 對象進行新的數(shù)據(jù)發(fā)送。

val ob = Observable.create(ObservableOnSubscribe<Int> { emitter ->
    (0..3).forEach {
        if (it == 2) emitter.onError(Throwable())
        else emitter.onNext(it)
    }
})

ob.onErrorResumeNext(Function<Throwable, ObservableSource<out Int>> { Observable.just(100) })
        .subscribe(observerInt)

發(fā)射了 0，1 后，會發(fā)射 onError，此時使用新的 Observable 發(fā)射 100，所以最終發(fā)射情況是：

onNext(0)
onNext(1)
onNext(100)
onComplete()

重載方法，參數(shù)直接傳發(fā)生錯誤時用的 Observable

ob.onErrorResumeNext(Observable.just(200,201)).subscribe(observerInt)

onNext(0)
onNext(1)
onNext(200)
onNext(201)
onComplete()

onErrorReturn/onErrorReturnItem

發(fā)生 onError 時返回一個特定的值，然后 onComplete。

ob.onErrorReturn { 8 }.subscribe(observerInt)

ob.onErrorReturnItem(8).subscribe(observerInt)

onNext(0)
onNext(1)
onNext(8)
onComplete()

onExceptionResumeNext

和 onErrorResumeNext 類似，區(qū)別在于如果 onError 傳的參數(shù)不是 Exception，那么會收到 onError。

只有一個構(gòu)造方法，參數(shù)是 Observable。

ob.onExceptionResumeNext(Observable.just(-1)).subscribe(observerInt)

onNext(0)
onNext(1)
onError()

修改發(fā)射 onError 時的參數(shù)。

val ob = Observable.create(ObservableOnSubscribe<Int> { emitter ->
   (0..3).forEach {
       if (it == 2) emitter.onError(Exception())
       else emitter.onNext(it)
   }
})

則同樣的代碼收到的是

onNext(0)
onNext(1)
onNext(-1)
onComplete()

retry/retryUntil/retryWhen

發(fā)生 onError 時會重試。

// 無限重試，其實最多是 Long.MAX_VALUE 次
public final Observable<T> retry()
// 指定重試次數(shù)
public final Observable<T> retry(long times)
// 參數(shù)回調(diào)返回 true 時重試
public final Observable<T> retry(Predicate<? super Throwable> predicate)
// 兩個控制條件都滿足時重試
public final Observable<T> retry(long times, Predicate<? super Throwable> predicate)
// 參數(shù)第一個泛型是重試次數(shù)，第二個是錯誤信息，返回 true 時重試
public final Observable<T> retry(BiPredicate<? super Integer, ? super Throwable> predicate)

// onNext(0)
// onNext(1)
// onNext(0)
// onNext(1)
// onNext(0)
// onNext(1)
// onError()
ob.retry(2).subscribe(observerInt)

// 一直 true，一直重試
ob.retry(Predicate { true }).subscribe{Log.e("RX", "$it")}

// 和 retry(2) 一樣效果
ob.retry(2, { true }).subscribe(observerInt)
ob.retry{ t1, _ -> t1 < 3 }.subscribe(observerInt)

var i = 0
ob.retryUntil {
    i++ > 1 // i > 1 時已經(jīng)重試兩次了，返回 true 以停止重試
}.subscribe(observerInt)

ob.retryWhen { 
  // 返回的 Observable 還活著就重試
  Observable.interval(0, 100, TimeUnit.MILLISECONDS).take(2)
}.subscribe{Log.e("RX", "$it")}