Extensionsの非同期周りの解説と自前実装2010-09-28最近id:okazukiさんが凄い勢いでRx解説を書いていて凄い!そして、刺激を受けますねー。いうわけで、今回はRxの非同期周りの解説をします。11「非同期処理用のSubject」ですって!早!私なんて一年かけて何も書いていやしなかったりするのに、この早さは本当に見習いたい、ごほごほ。そんなわけかで、色々と被ってしまっているのですが、Rxの非同期実行についてまとめます。FromAsyncPatternの3つ。Start,ToAsyncはFromAsyncPatternの簡易化みたいなものなので、実質は1つです。これらを使うと長さ1のIObservable<T>として非同期処理が扱えるわけです。Silverlightで実際に触って試してみてください。解説します......。Rxの非同期系処理は全てAsyncSubjectを通ります。AsyncSubjectのOnNextが非同期で実行された値の戻り値を、OnCompletedが非同期処理の完了を示します。通常はOnNextとOnCompletedはワンセットで非同期処理です。いうわけで、例えばSubscribeを二回押す→OnNextを一回押す→OnCompletedを一回押すで、非同期のIObservable<int>に対して二つSubscribe(Console.WriteLine)したということになり、1が二つ右側のログに表示されたはずです。続けてSubscribeを押す(非同期処理が完了した後にSubscribeした場合)とどうなるか?というと、1が追加されたはずです。ColdとしてIObservableの性質を少し取り上げました。ColdはSubscribeするとすぐに実行される。HotはSubscribeしても値が流れてくるまで待機する。AsyncSubjectは、つまり、両方の性質を持ちます。OnCompleted以前はHotで、値が流れてくる(非同期処理が完了する)まで待機される。OnCompleted以後はColdとなって、Subscribeするとすぐに値を流す。何となくAsyncSubjectの中身が想像付いてきました?そう、非同期の実行結果である値をキャッシュしています。複数回OnNextされたらどうなるか、というと、これは最後の値だけが残ります。(一度Resetしてから)OnNextを連打してからOnCompletedを押して、Subscribeしてみてください。非同期というとキャンセルはどうするの?というと、ありますよ!Subscribeの戻り値はIDisposable。イベントのRx化、FromEventの場合はイベントのデタッチでしたが、非同期で使う場合はキャンセルになります。Subscribeを押してから、Disposeを押して(キャンセル!)、OnNext→OnCompleted(非同期処理完了)を押してみてください。ログに何も出てきません。非同期処理が完了する前にDisposeしたということで、Subscribeがキャンセルされた、ということです。続けて(OnCompleted完了後)Subscribeを押すと......、ログに値が表示されます。Subscribeのキャンセルということですね。そうそう、ResetボタンはSubscribeやDispose、AsyncSubjectの状態をリセットして初期化します。Clearボタンはログの消去になります。AsyncSubjectの簡易実装AsyncSubjectの実態が大体分かったので、次は自分で実装してみましょう!その前にSubjectって何ぞや、というと、IObservable<T>かつIObserver<T>。これはRxネイティブのイベントだと思ってください。メソッド早見解説表をどうぞ。OnCompletedと同じ。}}あくまでこれは簡易化したものです。実際はこれより、もう少し複雑です。あとlockを省いているのにも注意。したのが知りたい場合はリフレクタとキャッキャウフフしてくださいということで。コードの中身ですが、Silverlightのデモで見てきた通りの、比較的単純な作りになっています。OnCompleted前後のSubscribeの挙動なんて、コードで見たほうが一目瞭然で早いですね。これの挙動がRxの非同期系の挙動です。AsyncSubjectではなくSubject(値のキャッシュなし)を使った場合は、非同期開始からSubscribeまでの間に完了してしまった場合、何も起きないことになってしまいます。キャッシュを取るというのは、非同期に最適な理に叶った、というか、少なくとも不都合は起きないような挙動になります。自前で非同期でIObservable<T>を返すようなメソッドを実装する場合は、必ずAsyncSubjectを使いましょう。ユーザーの期待する挙動を取らなければならないという、義務として。FromAsyncPatternの簡易実装ここまで来たら、ついでなのでFromAsyncPatternも実装してしまいましょう!AsyncSubjectがあれば簡単です。あ、こちらでも断っておくと、あくまで簡易実装であって実際のものとは若干異なります。EndInvokeに合わせてやるだけです。こんなの私もソラでは書けませんよー。今回は引数を一つ取るFromAsyncPatternを定義してみました。IObservable<TR>>なのは、引数を一つ取ってIObservable<T>にする、ということです。中身は割と簡単で、とりあえずBeginInvokeして、EndInvokeを待たずに(そりゃ非同期なので当然だ)とりあえずAsyncSubjectを戻します。これでIObservableとしてはHotの状態。OnCompletedを呼んでやる。これでSubscribeされていたものが実行されて、ついでにIObservableとしてColdになる。