Reactive Extensions for .NET (Rx) メソッド探訪第六回:exception handling

.NET Reactive FrameworkからReactive Extensions for .NET (Rx)に名称が変わったようなので、タイトルも変更。長いね。というわけで久しぶりなのですが、今回はざっとexception handling operators、つまり「Catch, Finally, Retry, OnErrorResumeNext」を見てみることにします。それとRun(ForEachなので説明不要ですが)。Rxって何？という人はHello, Reactive Extensionsをまず参照下さい。

Rxの花形はイベント合流系のメソッドにあると思うので、ひたすら脇役ばかりを紹介してちっとも本流に入ろうとしないのはどうかと思うのですけど、EnumerableExのCatchを見て、あー、こりゃ便利だ、ヤバい、便利だ、用途すぐ浮かんでしまった、というわけでしてCatchを紹介します。まずは、その浮かんだ例であるTwitterのタイムライン取得をどうぞ。例はIEnumerableに対してのものですが、IObservableに対してのものも同じです。

class Twitter
{
    public string Text { get; set; }
    public DateTime CreatedAt { get; set; }
}

static IEnumerable<Twitter> EnumerateUserTimeline(string userName)
{
    // {0}はユーザー名、{1}はページ番号 公開ユーザーのものを取得なら認証不要
    var format = "http://twitter.com/statuses/user_timeline/{0}.xml?page={1}";
    foreach (var page in Enumerable.Range(1, 1000))
    {
        var query = XDocument.Load(string.Format(format, userName, page))
            .Descendants("status")
            .Select(e => new Twitter
            {
                Text = e.Element("text").Value,
                CreatedAt = DateTime.ParseExact(e.Element("created_at").Value,
                    "ddd MMM dd HH:mm:ss zzzz yyyy",
                    CultureInfo.InvariantCulture,
                    DateTimeStyles.AssumeUniversal)
            });
        foreach (var item in query) yield return item;
    }
}

static void Main(string[] args)
{
    // 2009/11/23から今日までの投稿を古い順に並べるというもの
    var test = EnumerateUserTimeline("neuecc")
        .TakeWhile(t => t.CreatedAt >= new DateTime(2009, 11, 23))
        .OrderBy(t => t.CreatedAt)
        .ToArray();
    
    // これで基本的には問題ないわけですが、TwitterにはAPI制限があるので
    // ちゃんと全部取得出来るわけではなく、API制限発動 => 死亡になる可能性がある
    // 死んでもいいんだけど、せっかく取った死ぬ前のデータはがめておきたいよねえ
    // というわけで、そこで出番なのがRxのCatch！

    var test2 = EnumerateUserTimeline("neuecc")
        .TakeWhile(t => t.CreatedAt >= new DateTime(2009, 11, 23))
        .Catch((Exception e) => Enumerable.Empty<Twitter>())
        .OrderBy(t => t.CreatedAt)
        .ToArray();

    // 例外が発生したら握りつぶして、代わりにEnumerable.Emptyを返します
    // なので、例外発生前のデータは全て取得出来ています、素晴らしい！
}

といった感じです。つまりCatchは、そのまんまCatchです。Linqで全部書くのも良いんだけど、例外処理が出来なくてなあ、という不満がこれで解消されます。残りのFinally, Retry, OnErrorResumeNextですが、全部Catchの派生みたいなものです。とりあえず簡単な例を。

static IEnumerable<int> Iterate1To5()
{
    yield return 1;
    yield return 2;
    throw new DivideByZeroException(); // 嘘例外でも投げておく
    yield return 4;
    yield return 5;
}

static void Main(string[] args)
{
    // 1,2
    Iterate1To5().Catch((Exception e) => Enumerable.Empty<int>()).Run(Console.WriteLine);
    // 1,2,100,200
    Iterate1To5().Catch((Exception e) => new[] { 100, 200 }).Run(Console.WriteLine);
    // 1,2 -> 例外発生(ArgumentNullExceptionはDivideByZeroExceptionじゃないのでCatchしない)
    Iterate1To5().Catch((ArgumentNullException e) => new[] { 100, 200 }).Run(Console.WriteLine);
    // 1,2,100,200。つまりCatchの簡略版
    Iterate1To5().OnErrorResumeNext(new[] { 100, 200 }).Run(Console.WriteLine);
    // 1,2,Finally。これでtry-catch-finallyが出来あがる
    Iterate1To5()
        .Catch((Exception e) => Enumerable.Empty<int>())
        .Finally(() => Console.WriteLine("Finally"))
        .Run(Console.WriteLine);
    // 1,2 -> 1,2 -> 例外発生。例外を検知したら最初から列挙し直しての再試行
    // EnumerableExのRetryはバグっぽくてObservableとは違う動きをする
    // 明らかにオカシイのでそのうち修正されるでしょう
    Iterate1To5().ToObservable().Retry(2).Subscribe(Console.WriteLine);
}

最後に、中身をちゃんと知るには自分で実装するに限る、ということでIEnumerableでの拡張メソッドで再現してみました。Catchは本当に便利なので、わざわざRx使うのも、と思う場合は以下のコードを是非コピペして使ってくださいな。

// ループをぶん回すだけ、というもの(linq.jsではForce()が同様の働き)
public static void Run<TSource>(this IEnumerable<TSource> source)
{
    source.Run(_ => { });
}

// ようするにForEach
public static void Run<TSource>(this IEnumerable<TSource> source, Action<TSource> action)
{
    foreach (var item in source) action(item);
}

// try-catch句の中でyield returnが使えないので回りっくどいことに
public static IEnumerable<TSource> Catch<TSource, TException>(this IEnumerable<TSource> source,
    Func<TException, IEnumerable<TSource>> handler) where TException : Exception
{
    using (var enumerator = source.GetEnumerator())
    {
        while (true)
        {
            TException exception = null;
            var hasNext = false;
            try
            {
                hasNext = enumerator.MoveNext();
            }
            catch (Exception e)
            {
                exception = e as TException;
                if (exception == null) throw;
            }

            if (exception != null)
            {
                foreach (var item in handler(exception)) yield return item;
            }
            if (hasNext) yield return enumerator.Current;
            else yield break;
        }
    }
}

// Rxにはこういう、handlerがActionのオーバーロードが欲しいです
// わざわざ空のシーケンス投げるのは面倒くさいし、匿名型に対応できないじゃないか！
public static IEnumerable<TSource> Catch<TSource, TException>(this IEnumerable<TSource> source,
    Action<TException> handler) where TException : Exception
{
    return source.Catch((TException e) => { handler(e); return Enumerable.Empty<TSource>(); });
}

// OnErrorResumeNextはCatchの簡略版みたいなもんですね、別に必要ないような
public static IEnumerable<TSource> OnErrorResumeNext<TSource>(this IEnumerable<TSource> source, IEnumerable<TSource> next)
{
    return source.Catch((Exception e) => next);

}

// ToList().ForEach()とRun()ではactionの出るタイミングが変わることに注意
public static IEnumerable<TSource> Finally<TSource>(this IEnumerable<TSource> source, Action action)
{
    try { foreach (var item in source) yield return item; }
    finally { action(); }
}

// 本当は無限でやるべきなんでしょうが、int.MaxValueで。
public static IEnumerable<TSource> Retry<TSource>(this IEnumerable<TSource> source)
{
    return source.Retry(int.MaxValue);
}

// EnumerableExのRetryがバグ臭いのでObservable.Retryの挙動を採用しました
public static IEnumerable<TSource> Retry<TSource>(this IEnumerable<TSource> source, int retryCount)
{
    var count = 0;
    Exception exception = null;
    while (count < retryCount)
    {
        exception = null;
        foreach (var item in source.Catch((Exception e) => exception = e))
        {
            yield return item;
        }
        if (exception == null) yield break;
        count++;
    }
    throw exception;
}

どれもCatchの派生のようなものです、CatchイイよCatch。これは使いまくりたくなる。それにしてもtry-catchの中でyield returnが使えないのを、はじめて知りました。こんなことやろうとしたことがなかったので。あと、EnumerableEx.Retryはひっじょーにバグ臭いです。ちなみにEnumerableEx.Mergeもバグ臭い。全体的にEnumerableExはバグ臭さ全開です。明らかに(Observableから)適当に移植した感漂ってます。ヤバい。