なぜ Reactive Extension を使いたいのか?

ネットワークアクセスを含むリトライをこんなにシンプルに書ける環境他にあります?

IObservable<string> GetApiResultText(Func<WebRequest> webRequestFactory)
{
 return Observable.Defer(() => webRequestFactory().DownloadStringAsObservable())
 .Retry(3, TimeSpan.FromSeconds(3));
}

このメソッドはリクエストを生成するファクトリを放り込むと非同期でネットワークアクセスを行い、結果を文字列として取得し、エラーの際は3秒おきに計3回までリトライするものです。
意図即コード、僕にとってはこれがRxの一番の魅力です。
昔OptionとEitherを定義してC#で勝手if文無しバッチを書いた時もやっぱりRxの柔軟な機能が欲しくなってRxに寄せるはめになったんですよね。

また継承の多態を駆使したコードと組み合わせるとメソッドチェインによるフローの一覧性ってものすごく強力です。

そしたらもうこれはRxです、Rx。というわけでRxです。

RxFlow

シンプルなフロー制御機構です。RxのConnectableObservableと異なる形でシーケンスを分配します。

• 複数のシーケンスから同一のシーケンスに分岐
• 一つのシーケンスから複数のシーケンスへ分岐
• 複数のシーケンスから複数のシーケンスへ分岐
• 分岐から分岐へ
• 分岐の使いまわし

のような事が簡単にできるようになります。
Rx-Mainにしか依存していないので、Rx-Mainが使える環境では使えます。ちゃんとNugetもあります。

Branch

Branchクラス、これは分岐そのものを示します。CreateBranchのTでこの分岐に流れてくる値の型を指定し分岐を実現します。
作成したBranchは複数のシーケンスから参照することも、BranchからさらにBranchへ分岐することも可能です。

Junction

シーケンスをスイッチします。この例だと1~10までのシーケンスを生成し、2で割れるものはBranchAへ、3で割れるものはBranchBへ分岐させています。
分岐してbranchに流れていった値は後続に流れません。

static void Sample1()
{
 var branchA = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchA :" + i))
 .Subscribe());
 var branchB = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchB :" + i))
 .Subscribe());
 Observable.Range(1, 10)
 .Junction(i => i % 2 == 0, branchA)
 .Junction(i => i % 3 == 0, branchB)
 .Subscribe();
}
/*output
* branchA :2
* branchB :3
* branchA :4
* branchA :6 // Not processed in branchB
* branchA :8
* branchB :9
* branchA :10
*/

Distribution

Junctionの後続に値が流れる版です。2で割れるのでBranchAに流れた値が後続にも流れているため、3で割れる値が流れるBranchBにも流れて、結果6がそれぞれで出力されています。

static void Sample2()
{
 var branchA = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchA :" + i))
 .Subscribe());
 var branchB = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchB :" + i))
 .Subscribe());
 Observable.Range(1, 10)
 .Distribution(i => i % 2 == 0, branchA)
 .Distribution(i => i % 3 == 0, branchB)
 .Subscribe();
}
/*output
* branchA :2
* branchB :3
* branchA :4
* branchA :6 //processed in branchA
* branchB :6 //processed in branchB
* branchA :8
* branchB :9
* branchA :10
*/

複数シーケンスから同一Branchへの分岐例

static void Sample3()
{
 var branchA = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchA :" + i))
 .Subscribe());
 var branchB = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchB :" + i))
 .Subscribe());
 Observable.Range(1, 5)
 .Distribution(i => i % 2 == 0, branchA)
 .Distribution(i => i % 3 == 0, branchB)
 .Subscribe();
 Observable.Range(6, 5)
 .Distribution(i => i % 2 == 0, branchA)
 .Distribution(i => i % 3 == 0, branchB)
 .Subscribe();
}
/*output
* branchA :2
* branchB :3
* branchA :4
* branchA :6
* branchB :6
* branchA :8
* branchB :9
* branchA :10
*/

BranchからBranchへ分岐する例

static void Sample4()
{
 var branchA = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchA :" + i))
 .Subscribe());
 var branchB = Branch.CreateBranch<int>(input =>
 input.Do(i => Console.WriteLine("branchB :" + i))
 .Junction(i => i % 2 == 0, branchA)
 .Subscribe());
 Observable.Range(1, 10)
 .Junction(i => i % 3 == 0, branchB)
 .Subscribe();
}
/*output
branchB :3
branchB :6
branchA :6 //branchB to branchA
branchB :9
*/