はじめに

以前、UniRx オペレータ逆引きという記事は書いたのですが、正引きがなかったので改めてまとめました。

フィルタ系

名前	効果・用途	備考
`Where<T>`	1:OnNextメッセージの内容を条件式でフィルタリングする	後ろが`Select`の場合は最適化が自動的に走る
	2:OnNextメッセージの内容および発行回数でフィルタリングする
`OfType<T>`	指定した型にキャスト可能なメッセージのみ通す	`Cast`との違いは、変換失敗時に”何もしない”こと
`IgnoreElements<T>`	すべてのOnNextメッセージをフィルタリングする	何も通さない
`Distinct<T>`	過去に入力された同値のメッセージをフィルタリングする	`Func<TSource, TKey>`で比較値の取り出し方法の指定、および`IEqualityComparer`の指定が可能
`DistinctUntilChanged<T>`	直前に入力されたメッセージと同値の場合にフィルタリングする	`Func<TSource, TKey>`で比較値の取り出し方法の指定、および`IEqualityComparer`の指定が可能
`First<T>`	条件を満たした最初の1つのOnNextメッセージのみを取り出したあと、OnCompletedメッセージをセットで発行する	条件式を指定可能。OnNextメッセージが1つも入力されなかった場合は`OnError(InvalidOperationException)`が発行される。
`FirstOrDefault<T>`	条件を満たした最初の1つのOnNextメッセージのみを取り出したあと、OnCompletedメッセージをセットで発行する	条件式を指定可能。OnNextメッセージが1つも入力されなかった場合は`OnNext(default<T>)`が発行される。
`Last<T>`	条件を満たした最後の1つのOnNextメッセージのみを取り出したあと、OnCompletedメッセージをセットで発行する	条件式を指定可能。OnNextメッセージが1つも入力されなかった場合は`OnError(InvalidOperationException)`が発行される。
`LastOrDefault<T>`	条件を満たした最後の1つのOnNextメッセージのみを取り出したあと、OnCompletedメッセージをセットで発行する	条件式を指定可能。OnNextメッセージが1つも入力されなかった場合は`OnNext(default<T>)`が発行される。
`Single<T>`	条件を満たすOnNextメッセージは必ず1回入力されるという制約をつける	入力が0回だった、または2回以上条件を満たすOnNextメッセージが入力された場合は`OnError(InvalidOperationException)`が発行される。
`SingleOrDefault<T>`	条件を満たすOnNextメッセージは必ず1回入力されるという制約をつける	入力が0回だった、または2回以上条件を満たすOnNextメッセージが入力された場合は`OnNext(default<T>)`が発行される。
`Skip<T>`	1:先頭から指定した個数のOnNextメッセージを無視する	引数は`int`(無視する個数）
	2:購読開始から指定した期間メッセージを無視する	引数は`TimeSpan`および使用する`Scheduler`
`SkipWhile<T>`	条件を満たす間はメッセージを無視する。一度でも条件を満たさなくなるとそれ以降は常に通過させる。
`Take<T>`	1:先頭から指定した個数のOnNextメッセージのみ通過させ、その直後にOnCompletedメッセージを発行する	引数は`int`(通過させる個数）
	2:購読開始から指定した期間のみメッセージを通過させる。その後はOnCompletedメッセージを発行する	引数は`TimeSpan`および使用する`Scheduler`
`TakeWhile<T>`	条件を満たす間のみメッセージを通過させる。条件を満たさなくなるとOnCompletedメッセージを発行する
`TakeUntil<T>`	他の`Observable`からのOnNextメッセージ入力を受けた瞬間に、OnCompletedメッセージを発行する	`Observable`を外から止めたいときに使える
`TakeUntilDestroy<T>`	指定した`GameObject`および`Component`が破棄されたタイミングでOnCompletedメッセージを発行する	`GameObject`などに連動して`Observable`を止めたいときに使える
`TakeUntilDisable<T>`	指定した`GameObject`および`Component`が`OnDisable()`されたタイミングでOnCompletedメッセージを発行する
`TakeLast<T>`	1:OnCompletedメッセージが入力されたタイミングで、直前のOnNextメッセージを指定の個数分取り出す	OnCompletedメッセージが入力されるまで動作しない
	2:OnCompletedメッセージが入力されたタイミング、直前のOnNextメッセージを指定の期間分取り出す	OnCompletedメッセージが入力されるまで動作しない
`Throttle<T>`	短期間に大量にメッセージが入力された場合はそれを無視し、落ち着いたタイミングで最後の1つだけを取り出して流す	引数は`Timespan`（落ち着いたと判断するまでの猶予時間）
`ThrottleFrame<T>`	短期間に大量にメッセージが入力された場合はそれを無視し、落ち着いたタイミングで最後の1つだけを取り出して流す	引数はフレーム数
`ThrottleFirst<T>`	1度OnNextメッセージが入力されたらそれを流し、そのあとは一定時間メッセージを無視する	連打の防止などに使える。引数は`Timespan`
`ThrottleFirstFrame<T>`	1度OnNextメッセージが入力されたらそれを流し、そのあとは一定時間メッセージを無視する	連打の防止などに使える。引数はフレーム数

メッセージの変換系

名前	効果・用途	備考
`Select<T>`	1:OnNextメッセージの内容を別の値に変換する	後ろが`Where`の場合は最適化が自動的に走る
	2:OnNextメッセージおよび発行回数を用いて別の値に変換する
`Cast<T>`	OnNextメッセージを指定した型にキャストする	キャスト失敗時にはOnErrorメッセージが発行される
`AsUnitObservable`	OnNextメッセージの型を`Unit`型に変換する	`Select(_ => Unit.Default)`の省略記法
`AsSingleUnitObservable`	Observableが終了するタイミングで`Unit`型のOnNextメッセージを1回だけ発行する	`LastOrDefault().AsUnitObservable()`

メッセージやストリームの合成系

名前	効果・用途	備考
`Merge<T>`	1.複数のストリームの内容を並列に結合する	引数は`IObservable<T>[]`
	2.`IObservable<IObservable<T>>`を`IObservable<T>`にまとめる	`IObservable<IObservable<T>>`に対する拡張メソッド
	3.複数のストリームの内容を並列に結合する	`IEnumerable<IObservable<T>>`に対する拡張メソッド
`Concat<T>`	1.複数のストリームの内容を直列に結合する	引数は`IObservable<T>[]`
	2.`IObservable<IObservable<T>>`を先頭から順番に購読する	`IObservable<IObservable<T>>`に対する拡張メソッド
	3.複数の`IObservable<T>`を先頭から順番に購読する	`IEnumerable<IObservable<TSource>>`に対する拡張メソッド
`SelectMany<T>`	1.OnNextメッセージを使って新しい`Observable`を生成し、それを並列に合成する	`flatMap`に相当する
	2.OnNextメッセージを使って新しい`Observable`を生成し、そのメッセージを入力値と合成してから、並列に合成する	引数は「`Func<T, IObservable<TR>> collectionSelector`」「`Func<T, TC, TR> resultSelector`」の2つ
`ContinueWith<T>`	`SelectMany<T>`の軽量版。1回しか実行されない代わりに軽量。	1回しかメッセージが発行されない非同期処理に向く
`Switch<T>`	購読する対象の`Observable`を次々に切り替える	`IObservable<IObservable<T>>`に対してのみ使える
`Aggregate<T>`	OnNextメッセージに対して加工を行い、その結果を記録してさらに次のメッセージに引き継ぐ	`Scan<T>`との違いは、OnCompletedメッセージが発行されたタイミングで最終結果を出力する
`Scan<T>`	OnNextメッセージに対して加工を行い、その結果を記録してさらに次のメッセージに引き継ぐ	`Aggregate<T>`との違いは、こちらは1回処理するたびにメッセージを発行する
`Buffer<T>`	1.複数のOnNextメッセージを指定した個数分まとめて、1つのOnNextメッセージに加工する	`count` == `skip`
	2.複数のOnNextメッセージを指定した個数分まとめて、1つのOnNextメッセージに加工したあと、指定個数スキップする	`count`と`skip`を別々に設定できる。`Buffer(count:2, skip:1)`にすると直前のメッセージとセットにしてメッセージを発行できる。
	3.指定の時間間隔でメッセージをまとめる
	4.指定の時間間隔でメッセージをまとめたあと、指定した時間計測を一時中断する
	5.個数指定と時間指定を両方組み合わせてメッセージをまとめる	時間か個数どちらかの条件を満たしたタイミングで出力される
	6.メッセージをまとめ、外部からメッセージ入力されたタイミングで出力する	引数は`IObservable<TWindowBoundary>`。`TWindowBoundary`の型はなんでもよい
`BatchFrame<T>`	1.指定したフレーム数の間に発行されたOnNextメッセージを1つにまとめる	フレーム数と、`FrameCountType`を指定できる
	2.メッセージのタイミングを指定の`FrameCountType`のタイミング変換する	`IObservable<Unit>`の場合
`PairWise<T>`	1. 1個前のメッセージと最新のメッセージをセットにして出力する	`Buffer(count:2, skip:1)`とだいたいおなじ挙動
	2. 1個前のメッセージと最新のメッセージをセットにして、それをさらに加工して出力する	`Buffer(count:2, skip:1).Select()`に似ている
`Zip<TLeft,TRight,TResult>`	1. 型が異なる2つのストリームを購読し、値がそれぞれセットで揃ったタイミングで加工して出力する	`IObservable<TLeft>`に対して`IObservable<TRight>`を合成し、`IObservable<TResult>`になる
`Zip<T>`	2. 型が同じストリームを購読し、値がそれぞれセットで揃ったタイミングで加工して出力する。最大7ストリームまで合成できる	`IObservable<T>`に対する拡張メソッド
`Zip<T>`	3. 型が同じ複数のObservableをまとめ、`IList<T>`として出力する	「`IEnumerable<IObservable<T>>`に定義された拡張メソッド」または「`Observable.Zip`」から利用できる。合成できるストリーム数に上限はない
`ZipLatest<TLeft,TRight,TResult>`	1. 型が異なる2つのストリームを購読し、値がそれぞれセットで揃ったタイミングで最新値のみを取り出して加工して出力する	`Zip`との違いは、こちらは余剰に入力されたメッセージは破棄される
`ZipLatest<T>`	2. 型が同じストリームを購読し、値がそれぞれセットで揃ったタイミングで最新値のみを取り出して加工して出力する。最大7ストリームまで合成できる	`Zip`との違いは、こちらは余剰に入力されたメッセージは破棄される
`CombineLatest<TLeft,TRight,TResult>`	過去に入力されたメッセージ値を記憶し、それを流用して合成を強制的に行う	`Zip`との違いは、値がセットにならなくてもメッセージを無理やり合成する
`WithLatestFrom<TLeft,TRight,TResult>`	1つのストリームを主軸とし、そこに別のストリームの最新値を合成する	別のストリームのメッセージを、もう片方のストリームのタイミングで読み取る、といったことができる。例:`Update()`で発行されたメッセージを`FixedUpdate()`で読み取る
`Observable.WhenAll<T>`	複数のObservableがすべて終了状態になるのを待ち、完了時にその結果をまとめて通知する	`static`メソッド
`Amb<T>`	複数のストリームのうち、もっとも早くメッセージが到達したストリームをひとつだけ採択する

ストリームそのものを加工する

名前	効果・用途	備考
`Delay<T>`	指定した時間分、OnNextとOnCompletedメッセージを遅延させる	OnErrorは素通しする
`DelayFrame<T>`	指定したフレーム分、OnNextとOnCompletedメッセージを遅延させる	OnErrorは素通しする
`DefaultIfEmpty<T>`	1回もOnNextメッセージが入力されずにOnCompletedメッセージが入力された場合に、デフォルト値を発行する
`StartWith<T>`	1. 購読された瞬間に指定の値のOnNextメッセージを一番最初に発行する
`StartWith<T>`	2. 購読された瞬間に指定の関数を実行しその結果をOnNextメッセージとして一番最初に発行する	登録した関数は購読された瞬間に実行される
`StartWith<T>`	3. 購読された瞬間に指定された複数の値を一番最初に発行する	指定された値はそれぞれ個別のOnNextメッセージとして発行される
`GroupBy<TSource, TKey>`	1. OnNextメッセージを指定した条件でグループ分けを行い、それぞれに分割したストリームとして出力する	戻り値が`IObservable<IGroupedObservable<TKey, TSource>>`型になる
`GroupBy<TSource, TKey, TElement>`	2. OnNextメッセージを指定した条件でグループ分けを行い、それぞれに分割したストリームにした上でそれを変換してから出力する	戻り値が`IObservable<IGroupedObservable<TKey, TElement>>`型になる
`Repeat<T>`	OnCompletedメッセージが入力されると、自分より上流のストリームに対して再購読を実行する	ストリームが完了したときに`Subscribe()`を再実行してくれる。無限ループに注意
`RepeatSafe<T>`	`Repeat<T>`とほぼ同じ。ただし、OnCompletedメッセージが連続して入力されると処理を中断する	`Repeat<T>`を安全にしたもの
`RepeatUntilDestroy<T>`	指定した`GameObject`か`Component`が破棄されるまでの間、`Repeat<T>`として機能する	`Repeat<T>`を安全にしたもの
`RepeatUntilDisable<T>`	指定した`GameObject`か`Component`がDisableされるまでの間、`Repeat<T>`として機能する	`Repeat<T>`を安全にしたもの
`Sample<T>`	1. OnNextメッセージを指定の時間間隔でサンプリングする	一定時間ごとに、そのときの最新値を出力する
`Sample<T>`	2. OnNextメッセージを別のストリームの入力タイミングでサンプリングする	別ストリームから入力があるたびに、そのときの最新値を出力する
`SampleFrame<T>`	OnNextメッセージを指定のフレーム間隔でサンプリングする
`Timeout<T>`	1. OnNextメッセージの発行間隔が一定時間あいたらOnErrorメッセージを発行する
`Timeout<T>`	2. 指定の時刻までにストリームが完了しなかったらOnErrorメッセージを発行する
`TimeoutFrame<T>`	OnNextメッセージの発行間隔が一定フレーム数あいたらOnErrorメッセージを発行する
`Timestamp<T>`	OnNextメッセージにそのメッセージが発行された時刻を付与する	戻り値は`IObservable<Timestamped<T>>`
`TimeInterval<T>`	OnNextメッセージが発行された時間間隔を計測し、それをメッセージに付与する	戻り値は`IObservable<TimeInterval<T>>`
`FrameTimeInterval<T>`	OnNextメッセージが発行された時間間隔を計測し、それをメッセージに付与する。ただし時間の計測にUnityの`Time.time`または`Time.unscaledTime`を用いる	時間の計測方法が違う以外は`TimeInterval<T>`と同じ
`FrameInterval<T>`	直前のOnNextメッセージとの経過フレーム数をOnNextメッセージに付け加える
`Synchronize<T>`	メッセージ処理に排他ロックを行う	`lock`に用いるオブジェクトを複数ストリームで共有することもできる

エラーハンドリング

名前	効果・用途	備考
`Catch<T, TException>`	指定した型の例外を含むOnErrorメッセージがきた場合に登録した処理を実行し、任意のストリームに差し替える。型が一致しないOnErrorメッセージは無視する	引数の型は`Func<TException, IObservable<T>> errorHandler`。`TException`の型は必ず明示しないといけない
`Observable.Catch<T>`	指定された複数のストリームを先頭から順番に購読し、OnNextメッセージをそのまま伝える。途中でOnErrorメッセージが発生すると次のストリームにスイッチする。OnCompletedメッセージが発行されたタイミングで終了する	失敗したら次へ、失敗したら次へ、成功するまで繰り返す
`CatchIgnore<T, TException>`	指定した型の例外を含むOnErrorメッセージがきた場合に登録した処理を実行し、`Observable.Empty<T>`に差し替える。型が一致しないOnErrorメッセージは無視する	エラーが起きたときにもみ消して終了する
`Retry<T>`	OnErrorメッセージが発行されたときに、上流のストリームを再購読する	計何回まで試行するか指定できる
`OnErrorRetry<T>`	1. OnErrorメッセージが発行されたときに、上流のストリームを再購読する	引数を何も指定しない場合。`Retry<T>`と同じ挙動をする
`OnErrorRetry<T, TException>`	2. 指定した型の例外を含むOnErrorメッセージがきた場合に登録した処理を実行し、上流のストリームを再購読する	`Catch` + `Retry`の複合
`OnErrorRetry<T, TException>`	3. 指定した型の例外を含むOnErrorメッセージがきた場合に登録した処理を実行し、一定時間待ってから上流のストリームを再購読する	`Catch` + `Retry`の複合だが、リトライするまでのディレイをつけられる
`Finally<T>`	ストリームが解体されるタイミングで登録した関数を実行する	ストリームの解体とは次に挙げるシチュエーションのこと。「OnCompletedメッセージが発行される」「OnErrorメッセージが発行される」「Disposeにより購読が中断される」「Subscribeで登録した関数の処理途中で例外が発生する」

Hot変換用

名前	効果・用途	備考
`Multicast<T>`	指定の`ISubject`を用いてHot変換する
`Publish<T>`	1. `Subject<T>`を用いてHot変換する	`Multicast(new Subject<T>())`に相当
`Publish<T>`	2. `BehaviorSubject<T>`を用いてHot変換する	初期値を設定した場合はこちらの挙動になる。`Multicast(new BehaviorSubject<T>())`に相当
`PublishLast<T>`	`AsyncSubject<T>`を用いてHot変換する	`Multicast(new AsyncSubject<T>())`に相当。OnErrorメッセージもキャッシュしてしまうのでエラーハンドリングに注意
`Replay<T>`	`ReplaySubject<T>`を用いてHot変換する	`Multicast(new ReplaySubject<T>())`に相当
`RefCount<T>`	Hot変換時、自分を購読する`Observer`がいる場合にストリームを稼働させる。`Observer`がいなくなると自動で停止する	`IConnectableObservable<T>`に対してのみ利用可能
`Share<T>`	`Publish<T>().RefCount()`の省略記法

Schedulerの切り替え

名前	効果・用途	備考
`ObserveOn<T>`	メッセージの実行コンテキストを指定のSchedulerに切り替える
`ObserveOnMainThread<T>`	メッセージの実行コンテキストをUnityメインスレッドに切り替える	`MainThreadDispatchType`でメインスレッド上のどのタイミングにするかを指定できる
`SubscribeOn<T>`	ストリームの初期構築を指定のSchedulerで行う	あまり使うことはない
`SubscribeOnMainThread<T>`	ストリームの初期構築をUnityメインスレッド上で行う
`DelaySubscription<T>`	1. 指定された時間だけ待ってから`Subscribe()`を実行する	購読開始のタイミングをずらせる
`DelaySubscription<T>`	2. 指定された時刻になったら`Subscribe()`を実行する	購読開始のタイミングを指定できる
`DelayFrameSubscription<T>`	指定されたフレーム数だけ待ってから`Subscribe()`を実行する	UpdateやFixedUpdateのカウント数を指定できる

その他

名前	効果・用途	備考
`Do<T>`	ストリームのメッセージを用いて副作用を起こす。メッセージそのものは加工しない	ストリームの途中で、ストリーム外部の状態を変化させるときに使う
`DoOnError<T>`	OnErrorメッセージを用いて副作用を起こす。メッセージそのものは加工しない	OnErrorメッセージのみに反応
`DoOnCompleted<T>`	OnCompletedメッセージを用いて副作用を起こす。メッセージそのものは加工しない	OnCompletedメッセージのみに反応
`DoOnTerminate<T>`	OnErrorまたはOnCompletedメッセージが発行されたとき副作用を起こす。メッセージそのものは加工しない	OnErrorメッセージまたはOnCompletedメッセージに反応
`DoOnSubscribe<T>`	ストリームが`Subscribe()`されたときに副作用を起こす	購読された瞬間をログに出したりするとデバッグに便利
`DoOnCancel<T>`	ストリームが`Dispose()`によってキャンセルされたときに副作用を起こす	OnErrorメッセージやOnCompletedメッセージには反応しない
`ForEachAsync<T>`	非同期処理の結果を伝搬しつつ、そのメッセージ消費を行う	`Do().Last().AsUnitObservable()`に挙動は似ている。AsyncReactiveCommandと組み合わせると非常に便利
`Materialize<T>`	すべてのメッセージを`OnNext(Notification<T>)`メッセージへと変換する	OnErrorメッセージやOnCompletedメッセージをOnNextメッセージに格納してしまう。どんなメッセージが発行されたのかを列挙できるので、テストするときに便利
`Dematerialize<T>`	`Materialize<T>`によって変換されたメッセージをもとに戻す
`AsObservable<T>`	指定のオブジェクトのインタフェースを`IObservable<T>`に制限する	オブジェクトのダウンキャストやクロスキャストを防止できる
`ToArray<T>`	発行されたOnNextメッセージをすべてキャッシュし、OnCompletedメッセージが入力されたタイミングで1つの`OnNext(T[])`メッセージに変換して出力する
`ToList<T>`	発行されたOnNextメッセージをすべてキャッシュし、OnCompletedメッセージが入力されたタイミングで1つの`OnNext(IList<T>)`メッセージに変換して出力する
`Wait<T>`	ストリームが完了するのをスレッドをブロックして待機する	メインスレッドで使うとフリーズするので常用禁止