RxJSにおいて、onNextの処理中に発生した例外をキャッチするにはどうすれば良いのでしょうか？

Question

RxJSについて、おそらく大変初歩的な質問だと思うのですが、ご教授下さい。

ObservableをsubscribeしたときのonNext中の処理において例外が発生したとき、その例外をキャッチするにはどのように書けば良いのでしょうか？
現在は下記のように書いていますが、try/catchでもキャッチできず、catchErrorにもonErrorにも入らないため、subscribeが終了しない状態で固まってしまいます。

observable.pipe(
  catchError((error) => {
    console.log('catchError');
    throw error;
  })
).subscribe(
  // onNext
  (x) => {
    // 例外を投げるメソッド
    try {
      this.throwMethod();
    } catch (e) {
      throw e;
    }
  },
  // onError
  err => {
    console.log('onError');
    throw err;
  }
);

// 例外を投げるメソッド
throwMethod () {
  observable2.
  subscribe(
    // これが例外を投げる場合がある
    this.getObservable3()
    .pipe(
      catchError((err) => {
        throw(err);
      }
    )
    .subscribe(
      (x) => {
        // 正常時の処理
      },
     err => {
         throwError(err);
       }
     );
  )
}

どうぞよろしくお願い致します。

Comments

[kakajika]

try/catchでも例外が捕捉できないということは例外の投げ方が間違っているのではないでしょうか？throwMethodの中身も載せていただいたほうがいいと思います。あとcatch節の中で再度例外をthrowしているのは意味のないコードでは…？

Answer 1

Rxにおいては、例外の処理はObservableのチェーンによって後続の処理に渡していくのが原則です。以下のように、Observableを関数の戻り値として返し、処理をつなげていくように修正してみてください。

throwMethod = (): Observable<...> => {
  return observable2.pipe(
    flatMap(() => this.getObservable3())
  )
}

observable1.pipe(
  flatMap(() => throwMethod())
).subscribe(
  (x) => {
    // 正常時の処理
  },
  err => {
    // 例外処理
  }
);

Comments

[momoth]

アドバイス頂いたように、後続の処理で例外を受け取る方針で修正したところ、上手くonErrorで受け取ることができました！
ありがとうございました。

Answer 2

Rx.Observable.from([10, 20, 30])
    .subscribe(
        x => {
            console.log(x);
            try {
                throw new Error()
            }
            catch (e) { console.log(e) } // ここでちゃんとキャッチする
        });