RxJavaのエラーハンドリング

RxJavaでは非同期処理などで非常に便利なライブラリですが、そのエラーハンドリングもとてもスマートに行うことができます。ここではRxJavaのエラーハンドリングの詳細について説明したいと思います。

subscribeでのエラー処理

通常はsubscribe()でエラーを受け取ってハンドリングすることとなると思います。

subscribeでのエラー処理

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        // エラー時の処理をここで受け取る
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> error:java.lang.Throwable

注意が必要なのは下記のようにonNext()のハンドリングだけを行うようにしていると、onError()発火時にクラッシュします。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2

Crash!! 
by rx.exceptions.OnErrorNotImplementedException

のでエラーが発生する可能性がある場合はきちんとハンドリングしましょう。

Errorをキャッチする

onErrorReturn

Observableチェインのなかで発生したErrorをキャッチして、大体のObjectに変換することでsubscriberにErrorが渡されるのを防ぐことができます。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .onErrorReturn(new Func1<Throwable, String>() {
                    @Override
                    public String call(Throwable throwable) {
                        return "return";
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> on next: return
> completed

エラー時に無事return文字列に変換されました。

onErrorResumeNext

Observableチェインのなかで発生したErrorをキャッチして、その中で再度Observableを呼び出すことで、エラー時に代替のStreamを返すことができます。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .onErrorResumeNext(new Func1<Throwable, Observable<? extends String>>() {
                    @Override
                    public Observable<? extends String> call(Throwable throwable) {
                        return Observable.from(new String[]{ "resume 1", "resume 2"});
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> on next: resume 1
> on next: resume 2
> completed

エラー時にresume 1,resume 2のストリームに変換されました。

エラー時にリトライする

retry

retry()メソッドはErrorが起こった時に、自動的にsubscribeしなおしてくれる、非常に便利なメソッドです。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retry()
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2

...成功するまでリトライし続ける。。。

引数なしのretry()メソッドは、それまでの処理を成功するまでリトライし続けます。
これでは復旧できない問題が起こった時に無限ループになりかねないので、リトライの回数を制限するメソッドもあります。

// count: リトライ回数
public final Observable<T> retry(long count);

また、Errorの状況をみてリトライするかどうかも判断できます。

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retry(new Func2<Integer, Throwable, Boolean>() {
                    @Override
                    public Boolean call(Integer count, Throwable throwable) {
                    // ２回までは無条件にリトライする
                        if(count < 3){
                            return true;
                        }
                    // それ以降はIllegalStateExceptionの場合だけリトライする
                        return throwable instanceof IllegalStateException;
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> error:java.lang.Throwable

retryWhen

retryWhen()メソッドはちょっと複雑でわかりにくいですが、より細かくリトライ処理を制御するためのメソッドです。retryWhen()のコールバックは初回からリトライ後に起こるErrorをObservableとして受け取り、それを変換して何らかのObservableとして返します。そしてこのコールバックで返したObservableで何らかの値がonNext()されると、そのタイミングでリトライされます。また、このObservableでonComplete()もしくはonError()を呼び出されると、もとのObservableがonComplete()もしくはonError()終了します。

ちょっとわかりにくいので簡単に例を示します。

3秒後にリトライ

3秒後にリトライするには以下のように記述します。

エラー発生後3秒待ってからリトライする

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.timer(3, TimeUnit.SECONDS);
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後

> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後
> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後
> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後
> subscribe
> on next: emit 1
> on next: emit 2

....繰り返し

Errorのまま終了

Errorのまま終了するには

Errorのまま終了する

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.error(throwable);
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> error:java.lang.Throwable

このコードはちょうどretryWhenメソッドを適用せずに、Errorを受け取ったのと同じ動作となります。

ErrorをハンドリングせずにComplete

Errorをハンドリングせずにその場でCompleteするには一見以下のコードで良さそうですが、これでは実は正しく動作しません。

Errorが来たらempty()でcompleteしたい

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.empty();
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2

詳しくはここのやり取りにもあるのですが、ここでflatMapしてempty()で返しても、もともとObservable<Throwable>の流れをcompleteするわけではないので、ここでは無視されてしまいます。

実際、Errorが起こったらいきなりcompleteしたい処理を書きたい場合には、実際には以下のように、errorのstreamを最初の一つだけ受け取って、それを無視する、というコードを書く必要があります。難しいですね。

Errorが来たらempty()でcompleteしたい

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.take(1).ignoreElements();
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2
> completed

ちゃんとCompleteできた！！

ちなみにtake(0)だとerrorが起こる前にいきなり全体のObservableが終了してしまいます。

余談ですが、エラー時に何もせずcompletedにしたい場合はonErrorResumeNext()を使ったほうがスマートかも知れません。

         .onErrorResumeNext(new Func1<Throwable, Observable<? extends String>>() {
                    @Override
                    public Observable<? extends String> call(Throwable throwable) {
                        return Observable.empty();
                    }
                })

3回リトライしたら終了

上記の例をもとに3回リトライする場合は、

3回リトライ

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.take(3);
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

のように書きます。

> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> subscribe
> on next: emit 1
> on next: emit 2
> completed

ちゃんと3回リトライして終了しました。
この場合は先程のretry(count)のメソッドとの違いは、retry(count)ではリトライ回数が上限に達した後にerrorで終了するが、このケースではcompletedで終了する、という点にあります。

3秒後にリトライ、を3回試して終了

上記の例と、retryWhenの一番最初の例を組み合わせることで、3秒後にリトライ、を3回試して終了、という処理は以下のように書けます。

3秒後にリトライ、を3回試して終了

        Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        log("subscribe");
                        subscriber.onNext("emit 1");
                        subscriber.onNext("emit 2");
                        subscriber.onError(new Throwable());
                    }
                })
                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
                    @Override
                    public Observable<?> call(final Observable<? extends Throwable> observable) {
                        return observable.take(3).flatMap(new Func1<Throwable, Observable<?>>() {
                            @Override
                            public Observable<?> call(Throwable throwable) {
                                return Observable.timer(3, TimeUnit.SECONDS);
                            }
                        });
                    }
                })
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        log("completed");
                    }

                    @Override
                    public void onError(Throwable e) {
                        log("error:" + e);
                    }

                    @Override
                    public void onNext(String s) {
                        log("on next: " + s);
                    }
                });

> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後

> subscribe
> on next: emit 1
> on next: emit 2

// 3秒後

> subscribe
> on next: emit 1
> on next: emit 2
> completed

以上です。retryWhen()を駆使すればもっと細かなリトライ処理を書けそうですね！

あとがき

Advent Calendarの4日目としてエントリさせて頂いたのですが、いきなり期日に間に合わず、穴を開ける形となってしまいました。大変申し訳ありません。

このエントリを書く際に非常に困ったことがありました。初めはこのエントリのサンプルコードはすべてKotlinで書く予定だったんです。しかし、実際にためしてみるとKotlinではretryWhenのコードがコンパイル通らない！

以下のコードがコンパイルがとおらない

Observable
            .just("hoge")
            .retryWhen { it.flatMap{ Observable.timer(3, TimeUnit.SECONDS) } }

いろいろ調べたのですが、結局解決至らず、Kotlinでサンプルコードを書くのは諦めました。 orz