RX弱者がRXの基本の「き」を整理してみる

この記事は C# Advent Calendar 2018 21日目の記事です。

#はじめに

いつもコードを書いていて、
何か詰まるところがあると思っていたんですが
最近それがRXというもので、
自分がかなりのRX弱者ということに気づいたので
これを機に整理していこうと思います。

C#のアドベントカレンダーなので
C#のRXについて書こうと思ったのですが
普段さわっているものがUnityC#なので
具体的なものについては、
純粋なC#のRXではなくUniRXを書いています。

情報で間違っている部分があれば、
修正していただくか、教えていただけると助かります。

#RXとは？
・Reactive Extensionsの略
・GoFのデザインパターンの1つであるObserverパターンを実装するフレームワーク
・非同期 / イベント / 時間に関する処理をLINQで書ける

#Observerパターンとは？
Observer：観察
Subject(観察対象)の状態が変化すると、
Observer(観察者)に対して通知される。
状態変化に応じた処理を記述するときに使用される。
(ボタンを押したら○○する、攻撃を受けたらHPを減らす、など)

RXでは、
Subject=IObservable
Observer=IObserver

Observerパターンという名前だが、
実際にはSubject(IObserver)は
受動的に通知を待つだけなので
Publish-Subscribe(発行-購読)パターンとも呼ばれる。

#RXにおいて重要なもの
・ストリーム
・メッセージ
・購読
・オペレータ

###ストリーム
ストリーム：川、小川

プロバイダ、データソースとも呼ばれる。
イメージとしては、
ストリームという川がイベントを運び状態の通知を行う感じ。
時間軸があることがポイント。

###メッセージ

ストリームから流れてくる値、
IObservableが発行して、IObserbablerに伝える。

###購読

IObserbableがIObservableが発行した
メッセージを受け取って処理を行う。(Subscribe)
種類は以下の3つ

void OnNext(T value)→通常の処理を行う
void OnError(Exception ex)→発生した例外を受け取って処理を行う
void OnCompleted()→通知が終了したときの処理を行う

onComplete と onError のどちらかが発生した時点でストリームは終了する

###オペレータ
オペレータ：操作者

ストリームを加工したり(Whereとか)、作成したりする
メソッドチェーンで繋げて書くことができる

#基本のかたち
流れてきたメッセージをオペレータで濾過してSubscribeで処理

Observable./*オペレータ*/.Subscribe(//したい処理);

メッセージはたくさん流れてくるので、
オペレータで
残すものを選んだり（Where）
使いやすいように加工したりできます（Select）

##オペレータ

###Linqのかたちで使える
Where

//偶数のときだけ処理をする
Observable.Range(0,10)
        .Where(x=>x%2==0)
        .Subscribe({
            Console.WriteLine("偶数です");
        });

Select

//10倍にする（処理の中で変化させた値を使用できる）
Observable.Range(0,10)
        .Select(x=>x*10)
        .Subscribe({
            Console.WriteLine(x);
        });

Range()は引数で渡された数字を流すストリームを作ります。

##オペレータ以外
###Subject
C#RxではSubjectは異質な存在らしいですが、
UniRxでは基本の「き」らしいです。

特徴
・IObservableインターフェースとIObserverインターフェースを両方実装したクラス
Subject自体をsourceとして、
任意のタイミングでOnNext()OnError()OnCompleted()を呼び出せる
＝勝手にメッセージが流れてくるのではなく、
わざと外で変化させて値を流し込むもの

//入力

//登録
var s = new Subject<nt>();
s.Subscribe(
    x => Console.WriteLine("OnNext");
    ex => Console.WriteLine("OnError");
    () => Console.WriteLine("OnCompleted")
);

//発火
s.OnNext();
s.Completed();

//出力
OnNext
OnCompleted

https://qiita.com/acple@github/items/8d3a4d3414fa59adff70
https://qiita.com/ralph/items/f7205c8171826cc2153b
https://qiita.com/toRisouP/items/2f1643e344c741dd94f8

###時間/イベント/非同期
イベントと非同期については、
よく分からなかったので
ほぼ参考のみ載せました。

####時間
Observable.Timer()
発火タイミングを設定できる

//5秒後に１秒間隔で値が発行される
var t = Observable.Timer(TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(1));
var s = t.Subscribe(
    x => Console.WriteLine("OnNext"),
    ex => Console.WriteLine("OnError"),
    () => Console.WriteLine("OnCompleted")
);

5秒後に一秒間隔で
「OnNext」が表示され続け、
エラーがあった場合に「OnError」
s.Dispose()で「OnCompleted」

https://blog.xin9le.net/entry/2012/01/07/211152
https://blog.okazuki.jp/entry/20111106/1320584830
https://qiita.com/Marimoiro/items/a72b60315c797c19a27c

####イベント
難しかったので、調べたことと参考のみ。

FromEvent()でeventをObservableに変換する。

https://sakapon.wordpress.com/2016/10/28/reactive-drag/
http://any-programming.hatenablog.com/entry/2017/05/11/155410
https://befool.co.jp/blog/8823-scholar/unirx-from-event-args/
https://ufcpp.net/study/csharp/sp_event.html
https://qiita.com/Temarin/items/be5f9cea260580327700

###非同期
こちらも、難しかったので、調べたことと参考のみ。

Start,toAsync
発火させてから非同期で処理を開始できる。

https://blog.okazuki.jp/entry/20100928/1285645729
https://blog.xin9le.net/entry/2012/01/15/163827
http://light11.hatenadiary.com/entry/2018/11/07/225617

###疑問点
Repeat()は、ストリームを再生成するものらしいのですが
そうなるとRepeatがかかるたびに、
ストリームが増えていって
無駄なものが奥で走ることになるんじゃなかろうかと
思うのですが、いまだに分かっていません。
https://qiita.com/toRisouP/items/59d10ddec2e89b86600c

#他に覚えておきたいもの
###unit構造体
(rxに限らないので、あんまり関係なかった)
返り値が本当は必要ではないけど、
何らかの返り値を渡さなくてはいけないときに使うもの
voidの役割をしてくれる
通知のみしたいときなどに使う

var s=new Subject<Unit>();
s.Observable(Console.WriteLine);

s.OnNext("通知");

https://qiita.com/tamy0612/items/bfc4b034d94b7fee1b82
https://ufcpp.net/study/csharp/st_function.html
https://qiita.com/alucky0707/items/a677e5c9850aa765dd55
https://qiita.com/Temarin/items/107fa2e39f427e591dbd
https://qiita.com/toRisouP/items/851087b4c990d87641e6

#さいごに
基本の「き」をまとまってないですが、まとめてみました。
何となく概要がつかめたら後は書くだけ、
だと思っているので
あやふやな部分や空白は
書いてなじませていこうと思います。
読んでいただき、ありがとうございました。

#参考
https://qiita.com/acple@github/items/6cfee916f09632037a6e
http://introtorx.com/Content/v1.0.10621.0/02_KeyTypes.html#KeyTypes
https://blog.xin9le.net/entry/rx-intro
https://www.slideshare.net/okazuki0130/reactive-extensionsv01
https://techinfoofmicrosofttech.osscons.jp/index.php?Reactive%20Extensions%EF%BC%88Rx%EF%BC%89
https://speakerdeck.com/adarapata/bu-kunaireactive-extensions?slide=22
https://blog.okazuki.jp/entry/2015/03/23/203825
http://rxwiki.wikidot.com/marble-diagrams
http://wilfrem.github.io/learn_rx/operators.html
http://kmycode.hatenablog.jp/entry/2017/09/30/181312
https://qiita.com/hiruberuto/items/39e4126f470d8b84b291
https://tech.recruit-mp.co.jp/front-end/rxjs-intro/
https://liginc.co.jp/web/js/151272
https://qiita.com/Temarin/items/be5f9cea260580327700
https://qiita.com/Temarin/items/c2066394e91966c2d4ef
http://okapies.hateblo.jp/entry/2015/03/04/031148
http://tassi-yuzukko.hatenablog.com/entry/2018/06/04/150921
http://tc-kazuki.hatenablog.jp/entry/2017/12/16/160204
http://garicchi.hatenablog.jp/entry/2014/09/17/200000
http://noriok.hatenadiary.jp/entry/2018/08/25/191945
https://www.slideshare.net/torisoup/uni-rx
https://qiita.com/toRisouP/items/3cf1c9be3c37e7609a2f