狠狠撸

Reactive Extensionsで
WP7の非同期処理を簡単に

@neuecc – 2011/5/21

Profile
? Twitter
=> @neuecc
? Blog => http://neue.cc/
? HNはneuecc 読むときは“のいえ”と読ませてます
? ドメイン繋いだだけなので発音するの考えてなかっ
た(のでccは抜きで←発音しにくいですから)
? Microsoft MVP for Visual C#(2011/4-)

? WP7で作った物
? ReactiveOAuth, Utakotoha
? WP7の好きなtheme
? light + lime

Agenda
? LINQの概要/LINQとしてのReactive
Extensions
? 非同期処理の面倒さと如何にRxが癒すか
? .NETにおける非同期パターンの説明
? Rxの基本(購読, キャンセル, 例外処理)
? 非同期処理で使うRxのメソッド概略
? 作った物紹介

Language INtegrated Query
// クエリ構文
var query = from x in source
where x % 2 == 0
select x * x;

// メソッド構文
var query = source
.Where(x => x % 2 == 0)
.Select(x => x * x);

LINQって何？
? データソースを統一的な書法で処理できる
? WhereでフィルタしてSelectで射影できるならそれ
はLINQって言えます！
? SQL関係ないし、C#も飛び越えて生きる概念
? JavaScript移植もあるしね
? linq.js – http://linqjs.codeplex.com/
? RxJS(Reactive Extensions for JavaScript)

LINQのデータソースとは
to Objects to Xml to Sql
配列 XML Database
List<T> (JSON)
Stream
無限リスト

to Events to Asynchronous
TextChanged IO – WebRequest
ジェスチャー Timer – ポーリング
センサー Thread – 長時間かかる処理
MusicPlayer

Reactive Extensions

Reactive Extensions =
Linq to Events
Linq to Asynchronous
LINQにおけるデータソースの拡張
……というだけじゃない
時間という軸を中心にした基盤

Rxの基本軸は時間
IE<T> length

IE<T>

async
event

IO<T> time

=> IE<T>も乗っかることで「全てのデータソース」が合成可能に！

古き良き同期コード
var req = WebRequest.Create("http://hoge/");
var res = req.GetResponse();
var str = new StreamReader(res.GetResponseStream()).ReadToEnd();

? 簡単。でも、Silverlight/WP7には同期APIは無い。
? UIがブロックされるのダメ絶対
? Thread立ててそっちで実行させれば？
? まあそうです
? でもないものはないのでしょうがない
? そのかわり特に気を使わなくても必ずUIノンブロッ
キングになる(※但しCPUヘヴィな処理は除く)

しょうがないので非同期で書く
var req = WebRequest.Create("http://hoge");
req.BeginGetResponse(ar =>
{
var res = req.EndGetResponse(ar);
var str = new StreamReader(res.GetResponseStream())
.ReadToEnd();

Dispatcher.BeginInvoke(() => MessageBox.Show(str));
}, null);

-> EndHoge
? 非同期はBeginHoge
? 基本、クロージャ全開で書く
? 面倒くさいけれど、まあこれぐらいなら？

ネストするとかなりヤバい
{
var url = new StreamReader(res.GetResponseStream())
.ReadToEnd();
var req2 = WebRequest.Create(url);
req2.BeginGetResponse(ar2 =>
{
var res2 = req2.EndGetResponse(ar2);
var str = new StreamReader(res2.GetResponseStream())
.ReadToEnd();
}, null);
}, null);

通信箇所に例外処理は必須
? WP7ではネットワーク周りのコードでは100%例外
発生の可能性がある
? 圏外だったり通信が超低速だったりすると？
? Hello, Timeout.
? 山崎春のWebException祭り
? 何の手立てもしないとアプリ落ちるよ
? 予期される例外だし、固有の後処理もあるだろう
し、復帰可能にすべきなので、その場その場で
catchして始末するのが無難

内側なのは見た目だけ
{ 非同期中に起こる例外はEnd時に戻される
try
{
var url = new StreamReader(res.GetResponseStream()).ReadToEnd();
req2.BeginGetResponse(ar2 => ここの例外をcatchしてない
{
var str = new StreamReader(res2.GetResponseStream()).ReadToEnd();
Dispatcher.BeginInvoke(() => textBlock1.Text = str);
}, null);
} catchできるのは同じ関数のブロック内だけ
catch(WebException e)
{
Dispatcher.BeginInvoke(() => MessageBox.Show(e.ToString()));
}
}, null);

もはやカオスすぎて頭痛い
{
try
{
var url = new StreamReader(res.GetResponseStream()).ReadToEnd();
req2.BeginGetResponse(ar2 =>
{
try
{
var str = new StreamReader(res2.GetResponseStream()).ReadToEnd();
}
catch (WebException e)
{
}
}, null);
}
catch (WebException e)
{
}
}, null);

Reactive Extensionsを使うと
ネストが消滅し完全フラット
拡張メソッド(後で説明します)

WebRequest.Create("http://hoge")
.GetResponseAsObservable()
.Select(res =>
new StreamReader(res.GetResponseStream()).ReadToEnd())
.SelectMany(s => WebRequest.Create(s).GetResponseAsObservable())
.Select(res =>
new StreamReader(res.GetResponseStream()).ReadToEnd())
.ObserveOnDispatcher()
.Subscribe(
s => MessageBox.Show(s),
e => MessageBox.Show(e.ToString()));

内部で発生する例外は全てここで扱える

Rxを使うことの利点
? ネストがなくなって平らに
? 統一的な例外処理が可能

?+ その他の機能もいっぱい
? リトライ処理
? イベントやシーケンスとの合成など

非同期のもと
? 非同期パターンは概ね二つ
? APM(Asynchronous Programming Model)
? BeginXxx-EndXxx
? WebRequest.BeginGetResponseとか
? EAP(Event-based Asynchronous Pattern)
? XxxAsync-XxxCompleted
? WebClient.DownloadStringAsync/Copletedとか

? 将来的には？
? Rx(WP7では標準搭載ですが.NET4ではまだ)
? Task(.NET4では標準搭載ですがWP7ではまだ)
? C# 5.0 Async(まだCTP, 恐らく2年ぐらい先)

どっちがいいの？
? Rxで使うならAPMのほうが相性良い
? APMは上から下まで流れてるが、EAPは最後に発火
させなければならない
? これはネストする場合に致命的に面倒

// APM
.GetResponseAsObservable()
.Subscribe();
// EAP
Subscribe後に発火
var wc = new WebClient();
wc.DownloadStringCompletedAsObservable()
.Subscribe();
wc.DownloadStringAsync("http://hoge");

拡張メソッドのすゝめ
? APM,WebRequest->WebResponseはプリミティブ
すぎて、一々Stream扱ったり面倒くさい
? WebClientのDownloadString的なのが欲しい

? なら拡張メソッドで自作すれば解決
? FromEventやFromAsyncPatternは定型句なので、こ
ちらも拡張メソッドで隔離するのがお薦め

FromEvent(FromEventPattern)
戻り値はIEvent<EventArgs>のIO<T>

public static
IObservable<IEvent<DownloadStringCompletedEventArgs>>
DownloadStringCompletedAsObservable(this WebClient webClient)
{
return Observable.FromEvent<
DownloadStringCompletedEventHandler,
DownloadStringCompletedEventArgs>(
h => h.Invoke, // おまじない
h => webClient.DownloadStringCompleted += h,
h => webClient.DownloadStringCompleted -= h);
}

FromEvent<EventHandler,EventArgs>

FromAsyncPattern
? FromAsyncPatternの戻り値はデリゲート
? つまり自分でInvokeするまで実行されない
? 拡張メソッドにするなら即実行のほうが便利？
? Task.Factory.StartNew的なイメージで

public static IObservable<WebResponse>
GetResponseAsObservable(this WebRequest request)
{
return Observable.FromAsyncPattern<WebResponse>(
request.BeginGetResponse, request.EndGetResponse)
.Invoke();
}

DownloadStringAsync(の自作)
? Stringが戻ったほうが便利ですよね
? POSTなども同じように作っておくと楽になる
public static IObservable<string>
DownloadStringAsync(this WebRequest request)
{
return request.GetResponseAsObservable()
.Select(res =>
{
using (var stream = res.GetResponseStream())
using (var sr = new StreamReader(stream))
{
return sr.ReadToEnd();
}
});
}

最初の例もこんなにスッキリ
? Rxが提供してくれているのは基本的な道具
? Rxは「分離しやすい」のも特徴的なメリット
? <T>への拡張メソッドや、IO<T>->IO<T>の拡張メ
ソッドなどを作って、すっきりさせよう

.DownloadStringAsync()
.SelectMany(s => WebRequest.Create(s).DownloadStringAsync())
.ObserveOnDispatcher()
.Subscribe(
s => MessageBox.Show(s),
e => MessageBox.Show(e.ToString()));

書き方の基本
? System.Observableを参照する
? Microsoft.Phone.Reactiveを参照する
? WP7には標準搭載
? NET4版は別途インストールしてSystem.Reactiveを

? 定形的な流れは
FromEvent/FromAsyncして(IO<T>の生成)
SelectなりWhereなりLINQでデータを加工して
ObserveOnDispatcherして(値をUIスレッドに戻す)
Subscribeする

ところでキャンセルしたい
// Subscribeの戻り値はIDisposableなので、Disposeすればおk
var disposable = AsyncMethod().Subscribe();
disposable.Dispose();

// イベントの場合はデタッチになるよ
var buttonClick = button.ClickAsObservable().Subscribe();
buttonClick.Dispose();

// Listに入れてまとめてDisposeとか
var disposables = new List<IDisposable>();
disposable.Add(button.ClickAsObservable().Subscribe());
disposable.Add(button.ClickAsObservable().Subscribe());
disposable.ForEach(d => d.Dispose());

// IList<IDisposable>はCompositeDisposableというのもある
var disposables = new CompositeDisposable();
disposable.Dispose();

例外処理は？
? 何も書かない
? 例外はcatchせずスローされてくる
? SubscribeのonErrorに書く
? Rxのチェーンで発生した例外を全てcatchする
? ここに空のものを書けば例外無視が成立とかも
? ※Tips:一部メソッドが間に挟まれていると(Publishなど)
場合によってはcatchされなくなることも(Publishは内部
で自前でSubscribeしているためチェーンが途切れてる)
? Catchメソッドを使う
? 例外の型を指定した通常のtry-catchに近いもの
? 戻り値にEmptyを返せば終了、何か別の値を返せばそ
れを代替として流すということになる

基本的なのはこの3つ
// 何も書かないので例外がスローされてくる
.Subscribe(Debug.WriteLine);

// SubscribeのonErrorで全てcatch
.Subscribe(Debug.WriteLine, e => { });

// CatchでWebExceptionだけcatch
.Catch((WebException e) => Observable.Empty<string>())
.Subscribe(Debug.WriteLine);

SelectMany - 直列の結合

AsyncA AsyncB

Zip - 並列の結合

AsyncA

Result

AsyncB

SelectMany + Zip - 合成の例

AsyncA AsyncB

Result

AsyncC

AsyncA().SelectMany(a => AsyncB(a))
.Zip(AsyncC(), (b, c) => new { b, c });

ForkJoin - 並列同時実行

AsyncA

AsyncB
Result
AsyncC

AsyncD

Observable.ForkJoin(AsyncA(), AsyncB(), AsyncC(), AsyncD())
.Select(xs => new { a=xs[0], b=xs[1], c=xs[2], d=xs[3] });

多重蹿谤辞尘(厂别濒别肠迟惭补苍测)

AsyncA AsyncB AsyncC Result

var asyncQuery = from a in AsyncA()
from b in AsyncB(a)
from c in AsyncC(a, b)
select new { a, b, c };

まとめ
? FromEvent/Asyncは拡張メソッドで隔離
? ついでにReadToEndなんかも隔離
? とりあえずSubscribeの前にObserveOnDispatcher
? onErrorとCatchで例外をコントロール
? 色々な合成メソッドで流れをコントロール

? もう非同期なんて怖くない！
? Rxで書くと実は同期で書くよりも柔軟
? むしろもう同期でなんて書きたくない！

ReactiveOAuth
? http://reactiveoauth.codeplex.com/
? TwitterとかのOAuth認証/通信用ライブラリ
? 以下のWP7アプリで使われています！

? SongTweeter @iseebi
? NumberPush @mitsuba_tan
? Utakata TextPad @kaorun
? HirganaTwit @hatsune_

どんな時に使えるの？
? Twitterクライアントを作るわけではないけれど、
ステータスだけTwitterに投稿したい、などはよく
あること（特に最近は何でもTwitterだし）

new OAuthClient(ConsumerKey, ConsumerSecret, accessToken)
{
MethodType = MethodType.Post,
Url = "http://api.twitter.com/1/statuses/update.xml",
Parameters = { { "status", "ここに投稿する文章" } }
}
.GetResponseText() // 投稿して、戻り値を得る
.Select(XElement.Parse) // xmlの場合はパースしたいよね
.Subscribe(); // なにか処理するなり例外処理入れるなり

ポータビリティ
? WP7だけじゃなく.NET4/SL4向けもあり
? コードの99%をWP7と共有している
? 残り1%は最近の更新で.NET版Rxが一部WP7版と互換
なくなったため
? Rxをベースに置くと、コードの可搬性が圧倒的に
高まる

? 機能面でもRxにタダ乗り出来るので(エラー?リト
ライ?タイムアウトなど全部Rxにおまかせ)
? ライブラリ本体はシンプルなコードのままで強力
な機能を持てる

Utakotoha
? http://utakotoha.codeplex.com/
? 再生中の曲に応じて日本語歌詞を表示する
? マーケットプレイスで公開中(Free)
? ソースコードも公開中

コードの中身
? コードは無理やり全部Rxで割と実験的
? GUI周りは強引でボロボロで酷い
? イベント周りなどは面白く仕上がったかも

? Modelを別に立てた.NET4クラスライブラリにリン
クで参照することによりMSTestでユニットテスト
? Molesというモックライブラリを使ってWP7のイベ
ント自体を乗っ取り(音楽再生の情報をテストのた
めに任意に生成したり)

? 上手くいってるかはノーコメント

公式見るのがいいよやっぱり
? Data Developer Center - Rx
? http://msdn.microsoft.com/en-us/data/gg577609

? 二つのドキュメントが出ています
? Hands-on-Lab
? チュートリアル式で触りながら分かりやすく
? まず見て欲しい

? Design Guidelines
? マニュアルみたいなもの
? 更に深くへ

日本語情報は？
? ねぇよんなもん
? うちのサイトでも見てください:)
? http://neue.cc/
? http://neue.cc/category/rx

狠狠撸

Reactive Extensionsで非同期処理を簡単に

Recommended

More Related Content

What's hot (20)

Viewers also liked (6)

Similar to Reactive Extensionsで非同期処理を簡単に (20)

More from Yoshifumi Kawai (20)

Recently uploaded (6)

Reactive Extensionsで非同期処理を簡単に