Rx + WCF RIA Services = 簡単?? via(非同期プログラミングは辛いよ)

id:wave1008さんが非同期プログラミングは辛いよという記事を書かれていますが、このように非同期プログラミングは苦痛を伴うプログラミングです。

しかし、id:neueccさんがコメントで書かれていますが、現時点で便利なライブラリとしてReactive Extensionsというライブラリがあります。前に入門記事も書いてるので、Reactive Extensionsっていうのは何？？？っていう人は参照してみてください。

• Reactive Extensionsを始めてみた
• Reactive Extensions入門 1 Hello world
• Reactive Extensions入門 Hello worldその２
• Reactive Extensions入門 3「Where」
• Reactive Extensions入門 4「Observableを作成する便利なメソッド」
• Reactive Extensions入門 5「イベントを監視する」
• Reactive Extensions入門 6「非同期呼び出しの終わりを監視する」
• Reactive Extensions入門 7「Select」
• Reactive Extensions入門 8「バッファリング」
• Reactive Extensions入門 9「合成(MergeとZip)」
• Reactive Extensions入門 10「合成（SelectMany）」
• Reactive Extensions入門 11「非同期処理用のSubject」
• Reactive Extensions入門 12「非同期実行のためのメソッド」
• Reactive Extensions入門 13「値を集めて回る」
• Reactive Extensions入門 14 「OKが出るまで待つ」

因みに、id:griefworkerさんもReactive Extensionsをやりはじめたみたいなので、これから一緒に学んでいきたいという人は一緒に見てみるといいと思います。

• Reactive Extensions 入門(1)

さて、前置きが長かったですがid:wave1008さんのBlogの記事にあったように、顧客リスト→仕入先リスト→受注リストの順番で読み込む処理をやってみようと思います。

まず、WCF RIA ServicesはReactive Extensionsとの接点がデフォルトだと無い状態なので、間をつなぐ拡張メソッドを作成します。

public static class DomainContextExtensions
{
    public static IObservable<IEnumerable<TEntity>> LoadAsObservable<TEntity>(this DomainContext self,
        EntityQuery<TEntity> q)
        where TEntity : Entity
    {
        // Loadが終わったら通知するObservableを返す
        var subject = new AsyncSubject<IEnumerable<TEntity>>();
        self.Load(q, op =>
            {
                subject.OnNext(op.Entities);
                subject.OnCompleted();
            }, null);
        return subject.AsObservable();
    }
}

AsyncSubjectというものを使うことで、非同期処理の完了とともに監視してる人へ通知をするというようなことができる仕組みになってます。詳しくは以下の記事を見てください。

• Reactive Extensions入門 11「非同期処理用のSubject」

下準備が出来たのでメインの処理を書いていきます。

private EduDomainContext ctx = new EduDomainContext();

public void Load()
{
    this.ctx.LoadAsObservable(
        this.ctx.GetCustomersQuery())
        // 顧客を読み込んで
        .SelectMany(custmers =>
            {
                this.Customers.Source = custmers;
                return ctx.LoadAsObservable(ctx.GetSuppliersQuery());
            })
        // 仕入れ先を読み込んで
        .SelectMany(suppliers =>
            {
                this.Suppliers.Source = suppliers;
                return ctx.LoadAsObservable(ctx.GetOrdersQuery());
            })
        // 受注を読み込む
        .Subscribe(orders =>
            {
                this.Orders.Source = orders;
            });
}

どうですか？すごくシーケンシャルな処理になってるようにかけてると思います。因みにCustomersやSuppliersやOrdersはWCF RIA Services ToolkitのEntityListというクラスを使ってます。まぁ今回はそこが本題じゃなくてReactive Extensionsを使うことで、いかに非同期処理が直感的に書けるようになるかということを示したかったのですよ。

いい感じじゃないですか？