【Reactive Extensions】 IObservableの合成と分岐入門その1

みなさん、おはこんばんにちは。tsuchimotoです。
今回からReactive Extension（Rx）でも特に私が分かりづらいと思っているIObservableクラスの合成と分岐について説明したいと思います。

Reactive Extensionsでは、時間、イベント、非同期処理などをIObservableのシーケンス（イベントの流れ）として扱うことができます。
合成とは2本以上のイベントの流れを1本にまとめることです。
まずは分かりやすいメソッドから説明していきます。

Concatメソッド

Concatメソッドについて説明します。Concatメソッドは複数のIObservableのシーケンスを直列で繋いで1本の流れにします。

このメソッドで2本のシーケンスを1本にした場合、1本目のシーケンスがすべて完了してから2本目のシーケンスが開始されます。

3本のシーケンスを1本にした場合、1本目のシーケンスがすべて完了してから2本目のシーケンスが開始され、2本目のシーケンスがすべて完了してから3本目のシーケンスが開始されます。

Concatメソッドのシグネチャを以下になります。

// firstとsecondをつなぐ
public static IObservable<T> Concat<TSource>(this IObservable<TSource> first, IObservable<T> second
)

Concatメソッドの使用例を以下に示します。

Observable
    // 0,1,2と値を発行する
    .Range(0, 3).Select(i => "1st: " + i)
    .Concat(
        // 5,6,7,8,9と値を発行する
        Observable.Range(5, 5).Select(i => "2nd: " + i)
    )
    .Subscribe(s => Debug.Log("OnNext: " + s));

実行結果を以下に示します。
最初のシーケンス(1st)がすべて完了後に次のシーケンス(2nd)が開始されているのが分かると思います。

OnNext: 1st: 0
OnNext: 1st: 1
OnNext: 1st: 2
OnNext: 2nd: 5
OnNext: 2nd: 6
OnNext: 2nd: 7
OnNext: 2nd: 8
OnNext: 2nd: 9

実行結果を図にすると以下のようになります。

1st --0--1--2-|
2nd           -5--6--7--8--9|
r   --0--1--2--5--6--7--8--9|

Concatメソッドには以下のようなオーバーロードがあります。

// ①配列のすべての要素をつなぐ
public static IObservable<T> Concat<T>(IObservable<T>[] sources)
// ②IEnumerable<IObservable<T>>から取得できるすべての要素をつなぐ
public static IObservable<T> Concat<T>(this IEnumerable<IObservable<T>> sources)
// ③IObservable<IObservable<T>>から発行されるすべての要素をつなぐ
public static IObservable<T> Concat<TSource>(this IObservable<IObservable<T>> sources)
)

①のConcatメソッドの使用例を以下に示します。

Observable
    .Concat(
        // 0,1,2と値を発行する
        Observable.Range(0, 3).Select(i => "1st: " + i),
        // 5,6,7,8,9と値を発行する
        Observable.Range(5, 5).Select(i => "2nd: " + i)
    )
    // 講読
    .Subscribe(s => Debug.Log("OnNext: " + s));

実行結果は最初のコード例とまったく同じになります。

StartWithメソッド

StartWithメソッドについて説明します。
StartWithメソッドのシグネチャを以下に示します。

public static IObservable<T> StartWith<T>(this IObservable<T> source, params T[] values);

このメソッドはIObservableのシーケンスの先頭にvaluesで指定した要素を合成して1つのシーケンスにします。
このメソッドの使用例を以下に示します。

Observable
    // 1,2,3,4と値を発行するシーケンス
    .Range(1, 4)
    // 頭に10,20,30をつける
    .StartWith(10, 20, 30)
    // 購読
    .Subscribe(s => Debug.Log("OnNext: " + s));

実行結果を以下に示します。
StartWithで付けた値が初めに発行されているのが分かると思います。

OnNext: 10
OnNext: 20
OnNext: 30
OnNext: 1
OnNext: 2
OnNext: 3
OnNext: 4