2015/06/26 SelectManyでよかった件を追記
そういうケースありますよね？ たとえば
var source = ...何かIO<T>...; source .Select(async x => await HogeAsync(x)) ...省略...
みたいになったとき後続にはTask<T>が流れていきます。俺が欲しいのはTask<T>じゃなくてTの値なんだ！というときにはSelectManyメソッドを使うといい感じにTask<T>からTに変換してくれます。Switchメソッドを呼ぶといい感じにTask<T>からTに変換してくれます。
つまりこんな感じ。
var source = ...何かIO<T>...;
source
.SelectMany(x => HogeAsync(x))
.Subscribe(x =>
{
// xはHogeAsyncの戻り値のTask<T>のTの値
});
メモメモ。
話してきました。Reactive Extensionsの話しをするのはじめてなので緊張した。
www.slideshare.net
内容的には以下の記事に書いたことを話せたらなと思ってたのですが、思っていたように話せなかった。ちょっと心残り。
2記事続けて人の記事にのっかった記事になります。
解説は元記事がとても丁寧なのでそちらを一読することをお勧めします！C#固有の話とかあったらこちらで補足します。
1～10の数字を渡し、偶数だけにフィルタリングしたうえ、値を10倍にしてログ衆力するというプログラムです。
new[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 } .ToObservable() .Where(x => x % 2 == 0) .Select(x => x * 10) .Subscribe( x => Console.WriteLine(x), () => Console.WriteLine("completed"));
C#では、拡張メソッドがあるので配列からIObservableへの変換はToObservable拡張メソッドでやるところが特徴的です。
フィルタがWhere、値の加工がSelectという名前なところもC#のLINQの文化ですね。そして、Subscribeが、ラムダ式で行けるという点も見逃せません。これでかなりシンプルに書けます。
Rxのメソッドの基本的な実行順番を理解するためのプログラムです。
var sb = new StringBuilder(); Observable.Return(1) .Select(x => { sb.Append("1"); return x; }) .Subscribe( x => sb.Append("2"), // OnNext ex => sb.Append("3"), // OnError () => sb.Append("4")); // OnCompleted sb.Append("5"); Console.WriteLine(sb);
RxJava側のコードとの違いは、Java側はSubscribeでonNext, onCompleted, onErrorの順番でメソッドを定義していましたが、C#ではコメントにもある通りOnNext, OnError, OnCompletedの順番でラムダ式を指定します。それに伴い実行結果も以下のようにオリジナルとは異なります。
1245
Sleepを入れて動きを見てみるコードになります。
var sb = new StringBuilder(); Observable.Return(1) .Select(x => { Thread.Sleep(500); sb.Append("1"); return x; }) .Subscribe( x => sb.Append("2"), // OnNext ex => sb.Append("3"), // OnError () => sb.Append("4")); // OnCompleted sb.Append("5"); Console.WriteLine(sb);
実行結果もオリジナルと同じです。
1245
SubscribeOnの動作確認になります。SubscribeOnはSubscribeするスレッドを指定します。
var sb = new StringBuilder(); Observable.Return(1) .SubscribeOn(NewThreadScheduler.Default) .Select(x => { Thread.Sleep(500); sb.Append("1"); return x; }) .Subscribe( x => sb.Append("2"), // OnNext ex => sb.Append("3"), // OnError () => sb.Append("4")); // OnCompleted sb.Append("5"); Thread.Sleep(10000); Console.WriteLine(sb);
Schedulerは、Schedulersクラスから生えてる各種Schedulerへのインスタンスを取得するというのは非推奨になってるものが多いので、それっぽい名前のSchedulerのDefaultプロパティなどを使って指定します。今回は新しいスレッドを指定したかったので、NewThreadScheduler.Defaultを指定しています。
実行すると、500ms待ってからRxの中のストリームの処理が行われるのでオリジナル同様（OnCompletedの順番が違うのを除いて）になります。
5124
何処の処理がどのスレッドで実行するか確認するものになります。C#では、スレッド名が特についてないので、ManagedThreadIdを取得するようにしてみました。あとListに突っ込むのではなくてStringBuilderにAppendLineしてます。
var sb = new StringBuilder(); Observable.Return(1) .SubscribeOn(NewThreadScheduler.Default) .Select(x => { sb.AppendLine("1:" + Thread.CurrentThread.ManagedThreadId); return x; }) .Subscribe( x => sb.AppendLine("2:" + Thread.CurrentThread.ManagedThreadId), // OnNext ex => sb.AppendLine("3:" + Thread.CurrentThread.ManagedThreadId), // OnError () => sb.AppendLine("4:" + Thread.CurrentThread.ManagedThreadId)); // OnCompleted
実行すると以下のような結果になります。
1:3 2:3 4:3
スレッドのIDは実行環境によって変わると思います。
次にObserveOnを使ったコードのスレッドの切り替えをやります。RxJavaでは、mainスレッドに戻すSchedulerが提供されていますが、素のRxにはそのようなSchedulerがないのでC#のコードでは、SynchronizationContextを使用して実行スレッドをもとに戻しています。ここからはUIスレッドが存在することが前提となっているっぽいのでWPFのボタンクリックイベントハンドラにコードを書いています。
var sb = new StringBuilder(); Observable.Return(1) .SubscribeOn(NewThreadScheduler.Default) .Select(x => { sb.AppendLine("1:" + Thread.CurrentThread.ManagedThreadId); return x; }) .ObserveOn(SynchronizationContext.Current) .Subscribe( x => sb.AppendLine("2:" + Thread.CurrentThread.ManagedThreadId), // OnNext ex => sb.AppendLine("3:" + Thread.CurrentThread.ManagedThreadId), // OnError () => sb.AppendLine("4:" + Thread.CurrentThread.ManagedThreadId)); // OnCompleted await Task.Delay(5000); Debug.WriteLine(sb);
実行結果は以下のようになります。
1:12 2:10 4:10
因みにWPFやWinRTなどのようなUIスレッド上で処理を行うためのSchedulerは素のRxにはありません。ReactivePropertyなどのようなUI向けの機能を提供した別ライブラリなどで提供されています。
SubscribeOnを2回やったときの挙動の確認のプログラムになります。
var sb = new StringBuilder(); Observable.Return(1) .SubscribeOn(SynchronizationContext.Current) .Select(x => { sb.AppendLine("1:" + Thread.CurrentThread.ManagedThreadId); return x; }) .SubscribeOn(NewThreadScheduler.Default) .Select(x => { sb.AppendLine("2:" + Thread.CurrentThread.ManagedThreadId); return x; }) .ObserveOn(SynchronizationContext.Current) .Subscribe( x => sb.AppendLine("2:" + Thread.CurrentThread.ManagedThreadId), // OnNext ex => sb.AppendLine("3:" + Thread.CurrentThread.ManagedThreadId), // OnError () => sb.AppendLine("4:" + Thread.CurrentThread.ManagedThreadId)); // OnCompleted await Task.Delay(5000); Debug.WriteLine(sb);
実行結果は以下のようになります。最初にSubscribeOnしたスレッドで処理が実行されます。
1:9 2:9 2:9 4:9
最後は、途中からSubscribeOnした時の挙動です。
var sb = new StringBuilder(); Observable.Return(1) .Select(x => { sb.AppendLine("1:" + Thread.CurrentThread.ManagedThreadId); return x; }) .SubscribeOn(NewThreadScheduler.Default) .Select(x => { sb.AppendLine("2:" + Thread.CurrentThread.ManagedThreadId); return x; }) .Subscribe( x => sb.AppendLine("2:" + Thread.CurrentThread.ManagedThreadId), // OnNext ex => sb.AppendLine("3:" + Thread.CurrentThread.ManagedThreadId), // OnError () => sb.AppendLine("4:" + Thread.CurrentThread.ManagedThreadId)); // OnCompleted await Task.Delay(5000); Debug.WriteLine(sb);
オリジナルと同じで途中でSubscribeOnしても最初でしても同じ結果になってます。
1:11 2:11 2:11 4:11
ちょっとくらい違う動きするかと思ったらRxもRxJavaも同じ動きをしました。よく移植されてる。
最近Reactive Extensionsがバズってきてて、2年以上前に公開したReactive Extensionsのv1のメソッドを大体網羅したPDFが割と参照されてます。
www.slideshare.net
個人的な入門は、そこに全部書いたので今更感もありますが、Reactive Extensionsを今から入門する人向けに色々書いてみたいと思います。
IObservable<T>とIObserver<T>がコアのインターフェースになります。名前からわかるようにこれは、GoFのデザインパターンの1つ、オブザーバーパターンを表したインターフェースになります。
Observerパターンを簡単に説明すると、IObservableの状態に何かしら変化があったら、IObserverに変更通知を発行します。IObserverは、変更を受け取り何らかの処理を行います。
o: 状態変化
x: 処理
|: データの流れ
/: 終端
Observable ----o------o------o------o-------o----/
| | | | |
Observer ----x------x------x------x-------x----/
コードで表すと以下のようになります。
// Observable監視される側 class Observable : IObservable<string> { private List<IObserver<string>> obs = new List<IObserver<string>>(); public IDisposable Subscribe(IObserver<string> o) { this.obs.Add(o); return null; // 本当は購読解除するDisposableを返す } // 変更があったことを通知するメソッド public void Execute(string x) { foreach (var o in this.obs) { o.OnNext(x); } } }
// Observer監視する側 class Observer : IObserver<string> { // 変更通知があったときに呼ばれるメソッド public void OnNext(string x) { // とりあえず標準出力に出力するだけ Console.WriteLine(x); } public void OnCompleted() {} // no use public void OnError(Exception ex) {} // no use }
この2つを繋げます。
var observable = new Observable(); observable.Subscribe(new Observer()); // 監視する人を一人追加する observable.Execute("Hello"); // Hello と出力される observable.Subscribe(new Observer()); // 監視する人をもう一人追加する observable.Execute("World"); // Worldが2回出力される
図にすると大体以下のような感じです。
observable -----o(Hello)----------------------o(World)----------/
| |＼
observer1 -----x-----------------------------x-|---------------/
／
observer2 --------x-----------------/
Marble Diagramsという図を崩して書いたものになります。(AAだから許して！) この図は、右に行くと時間軸。縦方向が上がInputで下がOutputという非常に単純な図です。oとかでデータの発生を表します。
何故ObserverパターンをMarble Diagramsで描いてきたかというと、この図をイメージ出来るようになることが大事だからです。Reactive Extensionsの処理は、往々にして、このMarble Diagramsで描かれます。つまり時系列に応じて発生する値を入力として受け取って、何らかの処理を行い出力するのがReactive Extensionsでやっていることになります。
ここまでReactive Extensionsの記事なのにObserverパターンについて延々と語ってきました。ObserverパターンのObservableは、時間の経過に応じてデータを出力するものとみなすことができます。時間と入出力の関係を図示するMarble DiagramsでObserverパターンの処理の流れを描くことが出来たのはそのためです。 このように時間に応じてデータを発行するものをPush型のデータ、複数発行する場合はPush型のコレクションと言ったりします。
時系列に沿って何かしらの状態変化を起こして値を発行するものを、コレクションと見ることが出来たら後は、コレクション操作をするだけです。幸いにもC#にはLINQという強力なコレクション操作ライブラリがあります。このLINQをIObservableにまで拡張+αしたものがReactive Extensionsになります。(他言語にも大体コレクション操作ライブラリがあって、それに合うような形でObservableに対してコレクション操作を行うノリでReactive Extensionsがポーティングされています。)
一番よく使うLINQのメソッドのWhereやSelectを使ってReactive Extensionsのコード例を示します。1～10の値から、偶数のみを抽出して2倍した結果を出力してみます。Marble Diagramsで描くと以下のような処理です。
source --(1)--------(2)-------(3)-----(4)------(5)-----...
| |
Where -------------(2)---------------(4)--------------...
| |
Select -------------(4)---------------(16)-------------...
| |
処理 ------------(出力)-------------(出力)-----------...
var source = Observable.Range(1, 10); source .Where(x => x % 2 == 0) .Select(x => x * x) .Subscribe(x => Console.WriteLine("value = {0}", x));
実行すると以下のような結果になります。
value = 4 value = 16 value = 36 value = 64 value = 100
通常のLINQと違うのは、LINQがコレクションに入っている値を処理するのに対して、Reactive Extensionsは、sourceから発行された値を処理している点です。微妙ですが大事な点です。
これまでの例では、単純なコレクションの処理を示してきました。しかし、Reactive Extensionsは、時間の経過に伴って値を発行するものなら何でも処理できる点が強みです。このような特徴を持つものの代表例を以下に示します。
この中でもイベントを扱えるのは非常に強力で、イベントに対してフィルタリング、バッファリング、変換を行って処理をするといったことが簡単にできるようになります。
上記が個人的に思ってる基本的な内容です。応用的な内容としては以下のようなものがあります。簡単に触りだけ。
色々なものがReactive Extensionsで値を発行するソースになることを簡単に紹介しました。これらのデータは、Reactive Extensionsで提供されている様々な合成系のメソッドを使って合体させることが出来ます。
Reactive Extensionsで扱うデータは時間軸に応じて、値を発行するものだということは何度も言ってきました。その特徴から、Reactive Extensionsでは、時間に関する操作を行う処理もたくさん提供されています。
基本を押さえたうえで読むととても理解が捗ると思われる資料
大作。利用シーンが書かれてて素敵。
www.slideshare.net安定のぐらばくさん。マーブルダイアグラムとコードがならべて書いてあってとてもわかりやすい。
あと、上記資料からリンクされてるところも超参考になります。（リンク集め手抜き）
以上！良いRxライフを！
Reactive Extensionsはじめました というタイトルで超入門的な内容で話しをしようと思います。申し込みのほうは…公開と同時に80人枠が一気に埋まってしまうような感じで今からドキドキしてます。
生放送もされるっぽいので、そちらから見てもらえれば嬉しいです！
Reactive Extensionsには回数指定のリトライか、無限リトライしか用意されてません。世の中には、一定間隔をあけてリトライ処理をしたいという需要もあります。ということで、さくっと試してみました・
static class Ex { public static IObservable<T> RetryEx<T>(this IObservable<T> self, int retryCount, TimeSpan retrySpan) { int count = 0; IObservable<T> retry = null; retry = self.Catch((Exception ex) => { if (count++ < retryCount) { // retry return retry.DelaySubscription(retrySpan); } return Observable.Throw<T>(ex); }); return retry; } }
リトライ回数の以下ならDelaySubscriptionで、Subscribeを遅延させて、リトライの回数を超えていたら例外を投げています。
因みに、ReactivePropertyには、これをもっとしっかり作ったOnErrorRetryというメソッドが用意されてるので、お勧めしておきます。
Reactive Extensions 2.2 まとめ(1) - プログラミング勉強ノート
とてもよくまとまってます。ｆｍｆｍ。
これは別にUniRxじゃなくてもいいかな…。でも細かくUpdateを分割して書けるのは個人的に好きかも。
using UnityEngine; using System.Collections; using UniRx; public class MoveBehaviour : ObservableMonoBehaviour { public override void Awake() { // 加速度 var a = 5.0f; // 減速するスピード var downSpeed = 0.7f; // 最高速度 var maxSpeed = 10.0f; // 物体にかかってる力 var velocity = new Vector3(); // 速度を押されている水平キーに応じて加算していく this.UpdateAsObservable() .Select(_ => Input.GetAxis("Horizontal")) .Subscribe(x => { velocity.x += x * a * Time.deltaTime; velocity.x = Mathf.Min(velocity.x, maxSpeed); }); // 何も押されてないけど動いてるときは適当に減速する(右に動いてるとき) this.UpdateAsObservable() .Where(_ => velocity.x > 0) .Where(_ => !Input.GetButton("Horizontal")) .Subscribe(_ => { velocity.x -= downSpeed * a * Time.deltaTime; if (velocity.x < 0) { velocity.x = 0; } }); // 何も押されてないけど動いてるときは適当に減速する(左に動いてるとき) this.UpdateAsObservable() .Where(_ => velocity.x < 0) .Where(_ => !Input.GetButton("Horizontal")) .Subscribe(_ => { velocity.x += downSpeed * a * Time.deltaTime; if (velocity.x > 0) { velocity.x = 0; } }); // 力を物体にくわえる this.UpdateAsObservable() .Subscribe(_ => { this.rigidbody.velocity = velocity; }); base.Awake(); } }
これで水平方向になるから、あとはお好きな方向を同じ要領で追加していくだけですね。
またまたありがちな例ですがUniRxとUnityに慣れるために書いてみました。
UniRx版
using UnityEngine; using System.Collections; using UniRx; public class UniRxRollingBehaviour : ObservableMonoBehaviour { public override void Awake() { var drag = this.UpdateAsObservable() .Where(_ => Input.GetMouseButton(0)) .SkipUntil(this.UpdateAsObservable() .Where(_ => Input.GetMouseButtonDown(0)) .Select(_ => { RaycastHit rh; return Tuple.Create(Physics.Raycast(Camera.main.ScreenPointToRay(Input.mousePosition), out rh), rh); }) .Where(x => x.Item1 && x.Item2.collider.gameObject == this.gameObject)) .TakeUntil(this.UpdateAsObservable().Where(_ => Input.GetMouseButtonUp(0))) .Repeat() .Select(_ => Input.mousePosition); drag.Zip(drag.Skip(1), (x, y) => Tuple.Create(x, y)) .Select(x => x.Item2 - x.Item1) .Select(x => new Vector3(x.y, -x.x, x.z)) .Subscribe(x => this.transform.Rotate(x)); base.Awake(); } }
今回はがっつり1ステートメントに詰め込みました。そのぶん短くなったかな。コメントがあれば個人的には許容範囲かも。
続いて比較のためにUniRx無い版を書いてみた。状態管理のためのフラグとかが増えてきてちょっといやな感じを醸し出しはじめてるけど、まだ許容範囲。
using UnityEngine; using System.Collections; public class NormalRollingBehaviour : MonoBehaviour { private bool drag; private bool first = true; private Vector3 prevPosition; // Update is called once per frame void Update() { print("update"); // ドラッグ中じゃなくてマウスが押されたら if (!drag && Input.GetMouseButtonDown(0)) { // 対象オブジェクトの上でクリックされたかチェックして RaycastHit rh; if (Physics.Raycast(Camera.main.ScreenPointToRay(Input.mousePosition), out rh)) { if (rh.collider.gameObject == this.gameObject) { // 対象オブジェクトの場合は、ドラッグ中の状態にする drag = true; } } } // ドラッグ中にマウスが離されたらドラッグ中を解除して処理を抜ける if (drag && Input.GetMouseButtonUp(0)) { drag = false; return; } // ドラッグ中は if (drag) { print("drag"); if (!first) { var diff = Input.mousePosition - prevPosition; print(diff); this.transform.Rotate(new Vector3(diff.y, -diff.x, diff.z)); } first = false; prevPosition = Input.mousePosition; } } }
id:neueccさんの最新作と思われるUniRxをようやく触りました。
Reactive ExtensionsのUnity版。かっこいい。ということでマウスのドラッグを扱う簡単な例を書いて入門してみました。こんな感じ。
using UnityEngine; using System.Collections; using UniRx; public class UniRxCubeBehavior : ObservableMonoBehaviour { public override void Awake() { // ドラッグ対象のオブジェクト上でマウスが押されたときに値を発行するIObservable var mouseDownOnUniRxCube = this.UpdateAsObservable() .Where(_ => Input.GetMouseButtonDown(0)) .Select(_ => Camera.main.ScreenPointToRay(Input.mousePosition)) .Select(x => { RaycastHit rh; var hit = Physics.Raycast(x, out rh); return Tuple.Create(hit, rh); }) .Where(x => x.Item1 && x.Item2.collider.gameObject == this.gameObject) // マウスが離されたときに値を発行するIObservable var mouseUp = this.UpdateAsObservable() .Where(_ => Input.GetMouseButtonUp(0)); this.UpdateAsObservable() // マウスが押されてるとき .Where(_ => Input.GetMouseButton(0)) // 対象オブジェクトの上でマウスが押されるまでスキップして .SkipUntil(mouseDownOnUniRxCube) // マウスが離されるまで値を拾う .TakeUntil(mouseUp) // それを繰り返す .Repeat() // 現在のマウスの位置から移動位置を算出して .Select(_ => Input.mousePosition) .Select(x => { RaycastHit rh; Physics.Raycast(Camera.main.ScreenPointToRay(x), out rh, 1 << LayerMask.NameToLayer("Background")); return rh.point; }) // z座標は移動させないように変換して .Select(x => new Vector3(x.x, x.y, this.transform.position.z)) // 位置を設定する .Subscribe(x => this.transform.position = x); base.Awake(); } }
気合で1ステートメントに収めることもできるけど、個人的には適度にわけるのがお好み。これをCubeあたりに割り当てて、Backgroundというレイヤを割り当てた透明な背景を置いておくと、マウスでドラッグが動くようになります。
因みにRxを使わない場合はこんな感じになりました。
using UnityEngine; using System.Collections; public class NormalCubeBehavior : MonoBehaviour { private bool drag; void Update() { // ドラッグ中じゃなくてマウスが押されたら if (!drag && Input.GetMouseButtonDown(0)) { // 対象オブジェクトの上でクリックされたかチェックして RaycastHit rh; if (Physics.Raycast(Camera.main.ScreenPointToRay(Input.mousePosition), out rh)) { if (rh.collider.gameObject == this.gameObject) { // 対象オブジェクトの場合は、ドラッグ中の状態にする drag = true; } } } // ドラッグ中にマウスが離されたらドラッグ中を解除して処理を抜ける if (drag && Input.GetMouseButtonUp(0)) { drag = false; return; } // ドラッグ中は if (drag) { // 現在のマウスの位置を算出して RaycastHit rh; if (Physics.Raycast(Camera.main.ScreenPointToRay(Input.mousePosition), out rh, 1 << LayerMask.NameToLayer("Background"))) { // マウスの位置を新しい場所に設定する var newPos = new Vector3(rh.point.x, rh.point.y, this.transform.position.z); this.transform.position = newPos; } } } }
まだシンプルな例だから、そんなに変わりないかな。でも、ネストが嫌な感じを醸し出してる。
