Reactive Extensionsとエラーハンドリング

// 存在しないアドレスにアクセスしようとする(当然例外が起こる！)
// 出力結果は「リモート名が解決出来ませんでした」など。Timeoutを待つ必要はあります。
WebRequest.Create("http://goooooogllle.co.jp/")
    .DownloadStringAsync()
    .Subscribe(
        s => Console.WriteLine(s), // OnNext
        e => Console.WriteLine(e.Message), // OnError
        () => Console.WriteLine("completed!")); // OnCompleted
 
// OnNextのみの場合はcatchされずに例外が外にthrowされる
WebRequest.Create("http://goooooogllle.co.jp/")
    .DownloadStringAsync()
    .Subscribe(Console.WriteLine);

// TwitterのUserTimeLineは認証(OAuth)が必要なので例外が発生する！
// 出力結果は、以下のレスポンス
// {"error":"This method requires authentication.","request":"\/statuses\/user_timeline.json"}
WebRequest.Create("http://twitter.com/statuses/user_timeline.json")
    .DownloadStringAsync()
    .Catch((WebException e) => e.Response.DownloadStringAsync())
    .Subscribe(Console.WriteLine);

// 5が出たら例外出してみる
// 出力結果は1,2,3,4,completed
Observable.Range(1, 10)
    .Do(i => { if (i == 5) throw new Exception(); })
    .Catch((Exception e) => Observable.Empty<int>())
    .Subscribe(i => Console.WriteLine(i), e => { }, () => Console.WriteLine("completed"));

// Neverは何も返さない物
// Emptyとの違いは、OnCompletedすら発生しない
// 余談ですが、FromEventの第一引数は最新情報によると h => h.Invoke が一番短く書けてお薦めです！
var collection = new ObservableCollection<int>();
 
Observable.FromEvent<NotifyCollectionChangedEventHandler, NotifyCollectionChangedEventArgs>(
        h => h.Invoke, h => collection.CollectionChanged += h, h => collection.CollectionChanged -= h)
    .Select(e => (int)e.EventArgs.NewItems[0])
    .Do(i => { if (i == -1)  throw new Exception(); })
    .Catch((Exception e) => Observable.Never<int>())
    .Subscribe(Console.WriteLine, e => { }, () => Console.WriteLine("終了"));
 
// 出力結果は 300
collection.Add(300);  // 300が出力される
collection.Add(-1);   // 例外発生
collection.Add(1000); // デタッチ済みなので何も発生しない

// 出力結果は 例外発生, ふぁいなりー
Observable.Throw<int>(new Exception()) // Throwは例外のみを出すというもの
    .Finally(() => Console.WriteLine("ふぁいなりー"))
    .Subscribe(
        i => Console.WriteLine(i),
        e => Console.WriteLine("例外発生"),
        () => Console.WriteLine("こんぷりーてっど"));

// 1, 2, 3, ふぁいなりー1, 100, 101, 102, ふぁいなりー2
Observable.Range(1, 3)
    .Do(Console.WriteLine)
    .Finally(() => Console.WriteLine("ふぁいなりー1"))
    .TakeLast(1)
    .SelectMany(i => Observable.Range(100, 3))
    .Finally(() => Console.WriteLine("ふぁいなりー2"))
    .Subscribe(Console.WriteLine);

// 実行結果は「1, 2, 3」Catchは例外非発生時は後続を渡さない
Observable.Range(1, 3)
    .Catch(Observable.Range(100, 3))
    .Subscribe(Console.WriteLine);
 
// 実行結果は「1, 2, 3, 100, 101, 102」OnErrorResumeNextは例外非発生時も後続に繋ぐ
Observable.Range(1, 3)
    .OnErrorResumeNext(Observable.Range(100, 3))
    .Subscribe(Console.WriteLine);

// Access Start x3回のあとに401例外発生
Observable.Defer(() =>
    {
        Console.WriteLine("Access Start:");
        return WebRequest.Create("http://twitter.com/statuses/user_timeline.json").DownloadStringAsync();
    })
    .Retry(3)
    .Subscribe(Console.WriteLine);

// Timeoutは指定時間以内に値が通過しなければTimeoutExceptionが発生します
// カウントのタイミングはSubscribeされたらスタート
// この例では恐らく例外発生します(100ミリ秒で結果を返せる、ことは恐らくないでしょふ)
var wc = new WebClient();
Observable.FromEvent<DownloadStringCompletedEventHandler, DownloadStringCompletedEventArgs>(
        h => h.Invoke, h => wc.DownloadStringCompleted += h, h => wc.DownloadStringCompleted -= h)
    .Timeout(TimeSpan.FromMilliseconds(100)) // 100ミリ秒
    .Take(1) // これつけないとTimeoutのチェックが働き続けます←WebClientもWebRequest感覚になるようにしたほうがRx的にはお薦め
    .Subscribe(e => Console.WriteLine(e.EventArgs.Result));
 
wc.DownloadStringAsync(new Uri("http://bing.com/"));