Reactive Extensions入門 6「非同期呼び出しの終わりを監視する」

さて、前回はイベントをIObservableにする方法を見ました。次は、非同期呼び出しをIObservableにしてみようと思います。.NETでは、非同期の呼び出しはBegin〜, End〜のメソッドのペアでやるのが一般的です。

例えば・・・

// 5秒待って引数で渡された数だけ文字列返す
Func<int, string[]> f = i =>
    {
        Thread.Sleep(5000);
        return Enumerable
            .Range(0, i)
            .Select(a => "value : " + a)
            .ToArray();
    };

f.BeginInvoke(10, ar =>
    {
        var result = f.EndInvoke(ar);
        foreach (var s in result)
        {
            Console.WriteLine(s);
        }
    }, null);

こんな感じです。5秒たつと

value : 0
value : 1
value : 2
value : 3
value : 4
value : 5
value : 6
value : 7
value : 8
value : 9

と表示されます。まぁ、これはこれでいいんですけど、正直めんどくさい感じです。Reactive Extensionsを使うと、これがすっきりと書けるようになります！

Reactive ExtensionsではObservableクラスのFromAsyncPatternを使います。ということで、さっきのを書き換えてみます。

// 5秒待って引数で渡された数だけ文字列返す
Func<int, string[]> f = i =>
    {
        Thread.Sleep(5000);
        return Enumerable
            .Range(0, i)
            .Select(a => "value : " + a)
            .ToArray();
    };

// Begin〜, End〜のペアを渡すとIObservableを返すデリゲートを返す
Func<int, IObservable<string[]>> asyncFunc = Observable
    .FromAsyncPattern<int, string[]>(
    f.BeginInvoke, f.EndInvoke);

// FromAsyncPatternで返されたデリゲートを呼ぶとIObservableが返される
IObservable<string[]> ob = asyncFunc(10);

// Subscribeすれば結果を受け取れる
ob.Subscribe(arg =>
    {
        foreach (var s in arg)
        {
            Console.WriteLine(s);
        }
    });

// 5秒待たないと結果が返ってこないので待つ
Console.WriteLine("キーを押すと終了します");
Console.ReadKey();

実行結果は、以下のようになります。

キーを押すと終了します
value : 0
value : 1
value : 2
value : 3
value : 4
value : 5
value : 6
value : 7
value : 8
value : 9

因みに、説明用にメソッドの呼び出しを分割しましたが、繋げて書くとしたように書けます。

Observable.FromAsyncPattern<int, string[]>(
    f.BeginInvoke, f.EndInvoke)(10)
    .Subscribe(arg => arg.Run(Console.WriteLine));

すっきり！