RxJS の Operators (6) - Observable のエラーハンドリング

この記事は bouzuya's RxJS Advent Calendar 2015 の 8 日目かつ RxJS Advent Calendar 2015 の 8 日目です。

はじめに

今日は ReactiveX の Operators の Error Handling について RxJS の API ドキュメントやソースコードを見ていきます。

また RxJS 4.0.7 を対象にしています。

Observable のエラーハンドリング

subscribe (Observer) の onError

各種 Operator よりまず subscribe (Observer) の onError です。Observable.throw でエラーを流してみましょう。

import { Observable } from 'rx';

Observable
  .throw(new Error('ERROR!!'))
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onError: Error: ERROR!!

onError に入ると onCompleted は流れないようです。

では、途中で throw したらどうでしょう。

import { Observable } from 'rx';

Observable
  .from([1, 2, 3])
  .map(x => {
    if (x < 2) return x;
    throw new Error('x >= 2');
  })
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 1
// onError: Error: x >= 2

当然 try catch されて onError に流れてきます。

Observable.catch / Observable.prototype.catch

• ReactiveX - Catch operator
• Observable.catch API Document
• Observable.catch Source Code
• Observable.prototype.catch API Document
• Observable.prototype.catch Source Code

onError の際に代わりに流す Observable または Observable を返す関数を指定できます。

import { Observable } from 'rx';

Observable
  .from([1, 2, 3])
  .map(x => {
    if (x < 2) return x;
    throw new Error('x >= 2');
  })
  .catch(error => {
    console.log(`catch: ${error}`);
    return Observable.just(666);
  })
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 1
// catch: Error: x >= 2
// onNext: 666
// onCompleted

Observable.prototype.catch(handler) は Error を受けて続く observable を返します。

エラーをさらに流す場合は throw するか Observable.throw です。おそらく Observable.throw が行儀が良いのでしょう。

import { Observable } from 'rx';

Observable
  .from([1, 2, 3])
  .map(x => {
    if (x < 2) return x;
    throw new Error('x >= 2');
  })
  .catch(error => {
    console.log(`catch: ${error}`);
    return Observable.throw(error);
  })
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 1
// catch: Error: x >= 2
// onError: Error: x >= 2

Observable.prototype.catch(observable) は onError 時に代わりに流すものを指定するようですね。

import { Observable } from 'rx';

Observable
  .from([1, 2, 3])
  .map(x => {
    if (x < 2) return x;
    throw new Error('x >= 2');
  })
  .catch(Observable.just(666))
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// onNext: 1
// onNext: 666
// onCompleted

どちらの場合でも、元の Observable のエラーを無視して流し続けるのはちょっと難しそうですね。

Observable.prototype.retry

• ReactiveX - Retry operator
• Observable.prototype.retry API Document
• Observable.prototype.retry Source Code

onError の際に指定した回数だけリトライします。

import { Observable } from 'rx';

const maxRetryCount = 4;
let retryCount = 0;
Observable
  .from([1, 2, 3])
  .map(x => {
    console.log(`retryCount: ${retryCount}`);
    if (x >= 2 && retryCount < 2) {
      retryCount += 1;
      throw new Error('retryCount < 2');
    }
    return x;
  })
  .retry(maxRetryCount)
  .subscribe(
    value => console.log(`onNext: ${value}`),
    error => console.log(`onError: ${error}`),
    () => console.log('onCompleted')
  );
// retryCount: 0
// onNext: 1
// retryCount: 0
// retryCount: 1
// onNext: 1
// retryCount: 1
// retryCount: 2
// onNext: 1
// retryCount: 2
// onNext: 2
// retryCount: 2
// onNext: 3
// onCompleted

途中の onNext が流れる点、onError 以外であれば無視される点が注意点ですかね……。

おわりに

今日は Observable のエラーハンドリングを見ました。

これで本当にエラーハンドリングに十分なのか不安です。