RxJava:Error Handling Operatorを使ったエラーハンドリング方法を理解する

RxJavaではObservableでExceptionが発生した場合、デフォルトではそのObservableを監視しているObserverのonErrorメソッドが実行され、出力元のObservableが終了します。

想定外のエラーが発生して処理を続行することが困難な場合はonErrorで適切な後処理をしてやればよいですが、想定されたエラーの場合はエラーをハンドリングして別の処理に分岐させたいというケースもよくあるかと思います。

今回はRxJavaのonError以外のエラーハンドリングメソッドを試して、それぞれのメソッドごとの違いを理解することでエラーをより柔軟に扱う方法を学んでみます。

Catch Operator
参考

Catch Operator

RxJavaの元になっているReactiveXの公式ページではエラーハンドリング用のオペレータをError Handling Operatorと定義しています。Error Handling OperatorはCatchとRetryの2種類が定義されていますが、今回はCatch Operatorの使い方を調べてみます。

RxJavaのCatch OperatorとしてはonErrorReturn,onErrorResumeNext,onExceptionResumeNextの3種のオペレータが定義されています。以下でそれぞれの動作の違いを見てみましょう。

onError

それぞれのメソッドを確認する前にまずは通常のObservableでエラーが発生した場合の動作を見てみましょう。onErrorが実行されてその後onNext,onCompleteのいずれも実行されずに処理が終了することを確認します。

Observable.range(1, 10).map(i -> {
    if (i % 3 == 0) throw new RuntimeException();
    return i;
}).subscribe(
        i -> {
            Log.d("AAA", "onNext : " + i);
        },
        e -> {
            Log.e("AAA", "onError");
        },
        () -> {
            Log.d("AAA", "onComplete");
        }
);
// onNext : 1
// onNext : 2
// onError

onErrorReturn

onErrorReturnは出力元ObservableでのonErrorの発生を検知すると、ObserverのonErrorの代わりに実行されるメソッドです。onErrorReturnで返した値が出力元Observableの最後の出力となり、その値で最後のonNextが実行された後onCompleteが実行されます。

このオペレータを使うことで、「データ取得元で何らかのエラーが発生した場合でもエラー扱いにせず呼び出し元にデフォルト値を返す」という処理が簡単に実行できます。

Observable.range(1, 10).map(i -> {
    if (i % 3 == 0) throw new RuntimeException();
    return i;
}).onErrorReturn(e -> { // エラーを引数で受けている
    Log.e("AAA", "onErrorReturn");
    return -1;
}).subscribe(
        i -> {
            Log.d("AAA", "onNext : " + i);
        },
        e -> {
            Log.e("AAA", "onError");
        },
        () -> {
            Log.d("AAA", "onComplete");
        }
);
// onNext : 1
// onNext : 2
// onErrorReturn
// onNext : -1
// onComplete

onErrorResumeNext

onErrorResumeNextでは上記のonErrorReturnとは異なり、値ではなくObservableを返すことで出力元Observableでエラーが発生した際に、別のObservableに切り替えて引き続き値の出力を行うことができます。このメソッドでは出力元Observableで発生したエラーを引数に受けることができるので、エラー種別に応じて次に接続するObservableを変更するという処理が可能です。

Observable.range(1, 10).map(i -> {
    if (i % 3 == 0) throw new RuntimeException();
    return i;
}).onErrorResumeNext(e -> { // エラーを引数で受けている
    Log.e("AAA", "onErrorResumeNext");
    return Observable.range(10, 10);
}).subscribe(
        i -> {
            Log.d("AAA", "onNext : " + i);
        },
        e -> {
            Log.e("AAA", "onError");
        },
        () -> {
            Log.d("AAA", "onComplete");
        }
);
// onNext : 1
// onNext : 2
// onErrorResumeNext
// onNext : 10
// onNext : 11
// ~~~
// onNext : 19
// onComplete

onExceptionResumeNext

onExceptionResumeNextはonErrorResumeNextとよく似ていますが、onErrorResumeNextと異なりエラーの種別が判別できないという違いがあります。

onErrorResumeNextでは出力元Observableで発生したエラーを引数に受けることができますが、onExceptionResumeNextで引数にできるのは新しいObservableだけです*1。そのためonErrorResumeNextのようにエラー種別に応じて切り替え先のObservableを変更することはできません。

Observable.range(1, 10).map(i -> {
    if (i % 3 == 0) throw new RuntimeException();
    return i;
}).onExceptionResumeNext(Observable.range(10, 10) // Observableしか渡せない
).subscribe(
        i -> {
            Log.d("AAA", "onNext : " + i);
        },
        e -> {
            Log.e("AAA", "onError");
        },
        () -> {
            Log.d("AAA", "onComplete");
        }
);
// onNext : 1
// onNext : 2
// onNext : 10
// onNext : 11
// ~~~
// onNext : 19
// onComplete

参考

ReactiveX - Catch operator

*1:onErrorResumeNextでもObservableだけを引数に取るバージョンがあります。この場合、onExceptionResumeNextと同様の動きになります