Reactive Extensions再入門その１１「Catchメソッド」

過去記事インデックス

Reactive Extensions再入門番外編「センサー監視」

Catchメソッド

ここでは、Reactive Extensionsで例外処理を行うためのCatchメソッドについて紹介します。例外処理については、これまでもOnErrorで処理をするという方法をとってきましたが、Catchメソッドを使うことでエラーの際にリトライを行ったり、任意の値を後ろに対して流すといったことが出来ます。
Catchメソッドには、いくつかオーバーライドがありますが一番単純なものは、IObservable<T>を引数に渡すものになります。メソッドのシグネチャを下記に示します。

IObservable<TSource> Catch<TSource>(IObservable<TSource> o)

使用例を下記に示します。

var source = new Subject<string>();
source
    // sourceから例外が発生したらErrorという文字列を後続へ流す
    .Catch(Observable.Return("Error"))
    // 購読
    .Subscribe(
        s => Console.WriteLine("OnNext {0}", s),
        ex => Console.WriteLine("OnError {0}", ex),
        () => Console.WriteLine("OnCompleted"));

// 2つ値を発行したあと、例外を流して、その後にもう１つ値を発行する。
source.OnNext("A");
source.OnNext("B");
source.OnError(new Exception("例外"));
source.OnNext("C");

実行結果を下記に示します。

OnNext A
OnNext B
OnNext Error <- 例外が発生したのでErrorという文字列がSubscribeにわたってきている
OnCompleted <- その後、正常終了。

実行結果からもわかるとおり、CatchメソッドにObservable.Return(“Error”)を渡すことで、エラー時にはErrorという文字列を後続に対して流すようにしています。ここでは固定の値を返していますがIObservable<T>(この場合はIObservable)であれば何を返してもいいので、ここまでに紹介してきたメソッドを使って非同期処理を呼び出すことも、別のイベントの発火を待つことも可能です。
また、このCatchメソッドに渡すIObservableが終了したタイミングでOnCompletedが呼ばれることも上記サンプルから確認できます。そのためエラー時に必ず終了させるといった処理を行いたい場合は、空のIObservable<T>を返します。コード零を下記に示します。

var source = new Subject<string>();
source
    // エラーが起きたら終了する
    .Catch(Observable.Empty<string>())
    // 購読
    .Subscribe(
        s => Console.WriteLine("OnNext {0}", s),
        ex => Console.WriteLine("OnError {0}", ex),
        () => Console.WriteLine("OnCompleted"));

// 2つ値を発行したあと、例外を流して、その後にもう１つ値を発行する。
source.OnNext("A");
source.OnNext("B");
source.OnError(new Exception("例外"));
source.OnNext("C");

実行結果を下記に示します。

OnNext A
OnNext B
OnCompleted

Observable.Returnを使用した時とは違い、エラーが発生したタイミングでOnCompletedが呼ばれていることがわかります。
Catchメソッドには、上記以外のシグネチャのメソッド以外にFunc<TException, IObservable<TSource>>の形式のデリゲートを渡すオーバーライドがあります。このオーバーライドを使うと通常の例外処理のcatch句のように型指定で例外時の処理を振り分けることが出来ます。コード例を下記に示します。

var source = new Subject<string>();
source
    // ArgumentException発生時にはArgumentExceptionからの復帰という文字列を後続に流す
    .Catch((ArgumentException ex) => Observable.Return("ArgumentExceptionから復帰"))
    // NullReferenceException発生時には何もせず終了
    .Catch((NullReferenceException ex) => Observable.Empty<string>())
    // 購読
    .Subscribe(
        s => Console.WriteLine("OnNext {0}", s),
        ex => Console.WriteLine("OnError {0}", ex),
        () => Console.WriteLine("OnCompleted"));

// 2つ値を発行したあと、例外を流して、その後にもう１つ値を発行する。
source.OnNext("A");
source.OnNext("B");
source.OnError(new ArgumentException("例外")); // ## 1
// source.OnError(new NullReferenceException("例外")); // ## 2
// source.OnError(new Exception("例外")); // ## 3
source.OnNext("C");

上記の例では、2つのCatchメソッドでArgumentExceptionとNullReferenceExceptionの場合のエラー処理を行っています。ArgumentException発生時にはArgumentExceptionが発生した旨を文字列で後続に対して流すようにしています。NullReferenceExceptionの場合は、即座にシーケンスを終了するようにしています。コードの後半の## 1, ## 2, ## 3のコメントのある行のどれか１行だけ有効にした状態で実行することで、Catchメソッドの動作を確認できます。まず、## 1の行をコメントアウトしてArgumentExceptionを発生させた場合の実行例を下記に示します。

OnNext A
OnNext B
OnNext ArgumentExceptionから復帰
OnCompleted

期待したとおり、ArgumentExceptionから復帰という文字列がOnNextにわたってきていることが確認できます。次に、## 1の行をコメントにして## 2の行のコメントを外してNullReferenceExceptionを発生させたケースの実行例を下記に示します。

OnNext A
OnNext B
OnCompleted

こちらは、Observable.Empty<string>()を返しているため、例外が発生したタイミングでOnCompletedが呼ばれていることがわかります。最後に、## 3のコメントを外してExceptionを発生させたケースの実行例を下記に示します。

OnNext A
OnNext B
OnError System.Exception: 例外

この場合は、どのCatchメソッドでも処理されないためSubscribeのOnErrorまで例外が伝搬されています。このようにReactive Extensionsでは例外に対して複数の対処方法を用意しています。この中から、状況に応じて最適な例外処理選択して使用します。

かずきのBlog@hatena

すきな言語は C# + XAML の組み合わせ。Azure Functions も好き。最近は Go 言語勉強中。日本マイクロソフトで働いていますが、ここに書いていることは個人的なメモなので会社の公式見解ではありません。

Reactive Extensions再入門その１１「Catchメソッド」

過去記事インデックス

Catchメソッド