RxJavaとエラー処理

Posted on: 2019-10-18 2019-11-05
Tags: Library, RxJava

1前書き

この記事では、RxJavaを使用して例外とエラーを処理する方法について説明します。

まず、

Observable

は通常例外をスローしません。代わりに、デフォルトでは、

Observable

は

ObserverのonError（）

メソッドを呼び出して、回復不能なエラーが発生したことをオブザーバに通知し、それ以上

Observerの

メソッドを呼び出さずに終了します。

私たちが導入しようとしているエラー処理演算子は、

Observable

シーケンスを再開または再試行することによってデフォルトの振る舞いを変更します。

2 Mavenの依存関係

まず、

pom.xml

にRxJavaを追加しましょう。

<dependency>
    <groupId>io.reactivex.rxjava2</groupId>
    <artifactId>rxjava</artifactId>
    <version>2.1.3</version>
</dependency>

成果物の最新版はhttps://search.maven.org/classic/#search%7Cga%7C1%7Cg%3A%22io.reactivex.rxjava2%22%20AND%20a%3A%22rxjava%22[で見つけることができます。ここに]。

3エラー処理

エラーが発生したとき、我々は通常何らかの方法でそれを扱う必要があります。たとえば、関連する外部状態を変更してデフォルトの結果でシーケンスを再開するか、単にエラーが伝播するようにそのままにします。

3.1. エラーに対する処置

doOnError

があれば、必要なアクションを呼び出すことができます。エラー:

@Test
public void whenChangeStateOnError__thenErrorThrown() {
    TestObserver testObserver = new TestObserver();
    AtomicBoolean state = new AtomicBoolean(false);
    Observable
      .error(UNKNOWN__ERROR)
      .doOnError(throwable -> state.set(true))
      .subscribe(testObserver);

    testObserver.assertError(UNKNOWN__ERROR);
    testObserver.assertNotComplete();
    testObserver.assertNoValues();

    assertTrue("state should be changed", state.get());
}

アクションの実行中に例外がスローされた場合、RxJavaはhttp://reactivex.io/RxJava/2.x/javadoc/io/reactivex/exceptions/CompositeException.html[

CompositeException

]で例外をラップします。

@Test
public void whenExceptionOccurOnError__thenCompositeExceptionThrown() {
    TestObserver testObserver = new TestObserver();
    Observable
      .error(UNKNOWN__ERROR)
      .doOnError(throwable -> {
          throw new RuntimeException("unexcepted");
      })
      .subscribe(testObserver);

    testObserver.assertError(CompositeException.class);
    testObserver.assertNotComplete();
    testObserver.assertNoValues();
}

3.2. デフォルトのアイテムで再開する

ただし、

doOnError

を使用してアクションを呼び出すことはできますが、それでもエラーによって標準のシーケンスフローが中断されます。デフォルトのオプションでシーケンスを再開したい場合があります。http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Observable.html#onErrorReturnItem-T-[

onErrorReturnItem

]の動作は次のとおりです。

@Test
public void whenHandleOnErrorResumeItem__thenResumed(){
    TestObserver testObserver = new TestObserver();
    Observable
      .error(UNKNOWN__ERROR)
      .onErrorReturnItem("singleValue")
      .subscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertValueCount(1);
    testObserver.assertValue("singleValue");
}

動的デフォルトアイテムサプライヤが望ましい場合は、http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Observable.html#onErrorReturn-io.reactivex.functions.Function-[

onErrorReturn

]を使用できます:

@Test
public void whenHandleOnErrorReturn__thenResumed() {
    TestObserver testObserver = new TestObserver();
    Observable
      .error(UNKNOWN__ERROR)
      .onErrorReturn(Throwable::getMessage)
      .subscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertValueCount(1);
    testObserver.assertValue("unknown error");
}

3.3. 別のシーケンスから再開する

単一のアイテムにフォールバックする代わりに、http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Observable.html#onErrorResumeNext-io.reactivex.ObservableSource-[

onErrorResumeNext

を使用してフォールバックデータシーケンスを提供することもできます。]エラーが発生した場合これはエラーの伝播を防ぐのに役立ちます。

@Test
public void whenHandleOnErrorResume__thenResumed() {
    TestObserver testObserver = new TestObserver();
    Observable
      .error(UNKNOWN__ERROR)
      .onErrorResumeNext(Observable.just("one", "two"))
      .subscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertValueCount(2);
    testObserver.assertValues("one", "two");
}

フォールバックシーケンスが特定の例外タイプによって異なる場合、またはシーケンスを関数によって生成する必要がある場合は、関数を__onErrorResumeNextに渡します。

@Test
public void whenHandleOnErrorResumeFunc__thenResumed() {
    TestObserver testObserver = new TestObserver();
    Observable
      .error(UNKNOWN__ERROR)
      .onErrorResumeNext(throwable -> Observable
        .just(throwable.getMessage(), "nextValue"))
      .subscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertValueCount(2);
    testObserver.assertValues("unknown error", "nextValue");
}

3.4. 例外処理

のみ

RxJavaは、例外が発生した場合でもエラーが発生しなかった場合に、提供された

Observable

を使用してシーケンスを続行できるフォールバックメソッドも提供します。

@Test
public void whenHandleOnException__thenResumed() {
    TestObserver testObserver = new TestObserver();
    Observable
      .error(UNKNOWN__EXCEPTION)
      .onExceptionResumeNext(Observable.just("exceptionResumed"))
      .subscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertValueCount(1);
    testObserver.assertValue("exceptionResumed");
}

@Test
public void whenHandleOnException__thenNotResumed() {
    TestObserver testObserver = new TestObserver();
    Observable
      .error(UNKNOWN__ERROR)
      .onExceptionResumeNext(Observable.just("exceptionResumed"))
      .subscribe(testObserver);

    testObserver.assertError(UNKNOWN__ERROR);
    testObserver.assertNotComplete();
}

上記のコードが示すように、エラーが発生した場合、

onExceptionResumeNext

はシーケンスの再開を開始しません。

4エラー時に再試行

通常のシーケンスは、一時的なシステム障害またはバックエンドエラーによって壊れる可能性があります。このような状況では、シーケンスが修正されるまで再試行して待ちます。

幸いなことに、RxJavaにはそれを実行するオプションがあります。

4.1. リトライ

http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Observable.html#retry–

__,

を使用すると、

Observable__はエラーが発生しなくなるまで無限に再登録されます。しかし、ほとんどの場合、一定量の再試行を選択します。

@Test
public void whenRetryOnError__thenRetryConfirmed() {
    TestObserver testObserver = new TestObserver();
    AtomicInteger atomicCounter = new AtomicInteger(0);
    Observable
      .error(() -> {
          atomicCounter.incrementAndGet();
          return UNKNOWN__ERROR;
      })
      .retry(1)
      .subscribe(testObserver);

    testObserver.assertError(UNKNOWN__ERROR);
    testObserver.assertNotComplete();
    testObserver.assertNoValues();
    assertTrue("should try twice", atomicCounter.get() == 2);
}

4.2. 条件で再試行

条件付きの再試行は、http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Observable.html#retry-io.reactivex.functions.BiPredicate-[retry with predicate]またはを使用して、RxJavaでも実行可能です。

retryUntil

:

@Test
public void whenRetryConditionallyOnError__thenRetryConfirmed() {
    TestObserver testObserver = new TestObserver();
    AtomicInteger atomicCounter = new AtomicInteger(0);
    Observable
      .error(() -> {
          atomicCounter.incrementAndGet();
          return UNKNOWN__ERROR;
      })
      .retry((integer, throwable) -> integer < 4)
      .subscribe(testObserver);

    testObserver.assertError(UNKNOWN__ERROR);
    testObserver.assertNotComplete();
    testObserver.assertNoValues();
    assertTrue("should call 4 times", atomicCounter.get() == 4);
}

@Test
public void whenRetryUntilOnError__thenRetryConfirmed() {
    TestObserver testObserver = new TestObserver();
    AtomicInteger atomicCounter = new AtomicInteger(0);
    Observable
      .error(UNKNOWN__ERROR)
      .retryUntil(() -> atomicCounter.incrementAndGet() > 3)
      .subscribe(testObserver);
    testObserver.assertError(UNKNOWN__ERROR);
    testObserver.assertNotComplete();
    testObserver.assertNoValues();
    assertTrue("should call 4 times", atomicCounter.get() == 4);
}

4.3.

RetryWhen

これらの基本的なオプション以外にも、興味深い再試行方法があります。

__再試行

これは、ソースhttp://reactivex.io/RxJava/2.x/javadoc/io/reactivex/ObservableSource.html[

ObservableSource

]と同じ値を出力する

Observable、

say

“ NewO”、

を返します。 “ OldO”、

しかし返された

Observable

“ NewO”が

onComplete

または

onError

を呼び出すと、サブスクライバの

onComplete

または

onError__が呼び出されます。

また、

“ NewO”

がアイテムを発行すると、ソース

ObservableSource

“OldO”

への再購読がトリガされます。

以下のテストは、これがどのように機能するかを示しています。

@Test
public void whenRetryWhenOnError__thenRetryConfirmed() {
    TestObserver testObserver = new TestObserver();
    Exception noretryException = new Exception("don't retry");
    Observable
      .error(UNKNOWN__ERROR)
      .retryWhen(throwableObservable -> Observable.error(noretryException))
      .subscribe(testObserver);

    testObserver.assertError(noretryException);
    testObserver.assertNotComplete();
    testObserver.assertNoValues();
}

@Test
public void whenRetryWhenOnError__thenCompleted() {
    TestObserver testObserver = new TestObserver();
    AtomicInteger atomicCounter = new AtomicInteger(0);
    Observable
      .error(() -> {
        atomicCounter.incrementAndGet();
        return UNKNOWN__ERROR;
      })
      .retryWhen(throwableObservable -> Observable.empty())
      .subscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertNoValues();
    assertTrue("should not retry", atomicCounter.get()==0);
}

@Test
public void whenRetryWhenOnError__thenResubscribed() {
    TestObserver testObserver = new TestObserver();
    AtomicInteger atomicCounter = new AtomicInteger(0);
    Observable
      .error(() -> {
        atomicCounter.incrementAndGet();
        return UNKNOWN__ERROR;
      })
      .retryWhen(throwableObservable -> Observable.just("anything"))
      .subscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertNoValues();
    assertTrue("should retry once", atomicCounter.get()==1);
}

retryWhen

の典型的な使い方は、可変遅延を伴う制限付き再試行です。

@Test
public void whenRetryWhenForMultipleTimesOnError__thenResumed() {
    TestObserver testObserver = new TestObserver();
    long before = System.currentTimeMillis();
    Observable
      .error(UNKNOWN__ERROR)
      .retryWhen(throwableObservable -> throwableObservable
        .zipWith(Observable.range(1, 3), (throwable, integer) -> integer)
        .flatMap(integer -> Observable.timer(integer, TimeUnit.SECONDS)))
      .blockingSubscribe(testObserver);

    testObserver.assertNoErrors();
    testObserver.assertComplete();
    testObserver.assertNoValues();
    long secondsElapsed = (System.currentTimeMillis() - before)/1000;
    assertTrue("6 seconds should elapse",secondsElapsed == 6 );
}

このロジックが3回再試行し、各再試行を徐々に遅らせることに注目してください。

5概要

この記事では、RxJavaでエラーと例外を処理する方法をいくつか紹介しました。

エラー処理に関連するRxJava固有の例外もいくつかあります。詳しくは-them[オフィシャルウィキ]を参照してください。

いつものように、完全な実装はhttps://github.com/eugenp/tutorials/tree/master/rxjava-2[over Github]で見つけることができます。