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

みなさん、おはこんばんにちは。tsuchimotoです。

今回はUniRx（UnityのRxライブラリ）を使ったIObservableの合成と分岐入門の2回目です。

Mergeメソッド

Mergeメソッドについて説明します。Mergeメソッドは複数のIObservableシーケンスを1つにまとめます。

このメソッドで2つのシーケンスを1つにした場合、そのシーケンスの発行タイミングは合成前と変わりません。

つまり、Mergeメソッドは複数のシーケンスを合成後もシーケンスの発行タイミングを変えずに、そのまま1つのシーケンスにできます。

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

// firstとsecondを合成する
public static IObservable<T> Merge<T>(this IObservable<T> first, IObservable<T> second);

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

// a0,a1,a2と値を発行する
var s1 = Observable.Interval(TimeSpan.FromMilliseconds(250))
    .Take(3)
    .Select(i => "a" + i);
// b0,b1,b2,b3,b4と値を発行する
var s2 = Observable.Interval(TimeSpan.FromMilliseconds(150))
    .Take(5)
    .Select(i => "b" + i);

s1.Merge(s2).Subscribe(
    s => Debug.Log("OnNext: " + s),
    s => Debug.Log("Completed"));

実行結果を以下に示します。
合成前の2つのシーケンスの発行タイミングを保ったまま、1つのシーケンスにまとめられたことを示しています。

OnNext: b0
OnNext: a0
OnNext: b1
OnNext: a1
OnNext: b2
OnNext: b3
OnNext: a2
OnNext: b4

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

// ①配列の全てをマージ
public static IObservable<T> Merge<T>(params IObservable<T>[] sources);
// ②返されたIObservable<T>をすべてマージ
public static IObservable<T> Merge<T>(this IObservable<IObservable<T>> sources);

①は引数で渡したIObservableの配列を全てマージします。コード例を下記に示します。

// a0,a1,a2と値を発行する
var s1 = Observable.Interval(TimeSpan.FromMilliseconds(250))
    .Take(3)
    .Select(i => "a" + i);
// b0,b1,b2,b3,b4と値を発行する
var s2 = Observable.Interval(TimeSpan.FromMilliseconds(150))
    .Take(5)
    .Select(i => "b" + i);

Observable.Merge(s1, s2).Subscribe(
    s => Debug.Log("OnNext: " + s),
    s => Debug.Log("Completed"));

実行結果は先ほどと同じになります。

②は少々複雑になります。 IObservable<T>を返すようなIObservableに対して使用でき、返されたIObservable<T>をすべてマージします。

コード例を下記に示します。

var source = new Subject<string>();
source
    // IObservable<string>からIObservable<IObservable<string>>に変換
    .Select(i => Observable
        // 0,1,2と値を3つ発行
        .Interval(TimeSpan.FromMilliseconds(250)).Take(3)
        // 値を変換
        .Select(j => i + j))
    // IObservable<IObservable<T>>からIObservable<T>へマージ
    .Merge()
    // 購読
    .Subscribe(
        i => Debug.Log("OnNext: " +  i),
        i => Debug.Log("OnCompleted"));

// 値の発行
source.OnNext("a");
source.OnNext("b");
source.OnNext("c");
source.OnNext("d");

実行結果を以下に示します。

値を3回発行するIObservable<string>を4つマージしています。

よって、値は12回発行されます。

OnNext: a0
OnNext: b0
OnNext: c0
OnNext: d0
OnNext: a1
OnNext: b1
OnNext: c1
OnNext: d1
OnNext: a2
OnNext: b2
OnNext: c2
OnNext: d2

②の形が理解できれば、次回に説明するSelectManyも理解できると思います。