Merge

1. 概要

Merge は、複数の Observable シーケンスの要素を、届いた順に 1 つのシーケンスへ統合するオペレーターです。

各ソースに同時に購読し、どのソースから値が来てもそのまま下流へ流します。そのため、複数のソースが交互に値を発行した場合は、それらの値が 1 本のシーケンス上で混ざった順序になります。すべてのソースが完了したとき、結果のシーケンスも完了します。

2. シグネチャ

2 つの Observable をマージする（オペレーターオーバーロード）

public static Observable<T> Merge<T>(this Observable<T> source, Observable<T> second)

2 つの同一型 Observable を統合します。最も一般的な使い方です。

source.Merge(second)

可変個数の Observable をマージする（ファクトリメソッド）

public static Observable<T> Merge<T>(params Observable<T>[] sources)
public static Observable<T> Merge<T>(this IEnumerable<Observable<T>> sources)

任意の個数の Observable を同時に購読して統合します。

Observable.Merge(source1, source2, source3)

Observable の Observable をフラット化する

public static Observable<T> Merge<T>(this Observable<Observable<T>> sources)

外側の Observable が内側の Observable を発行するたびに購読を追加し、すべての内側のシーケンスの要素をフラットに統合します。

outerObservable.Merge()

overload の使い分け

overload	使う場面
Merge(Observable<T>)	2 つの Observable を統合する基本的なケース
Merge(params Observable<T>[])	3 つ以上の Observable をまとめて統合する場合
Merge(IEnumerable<Observable<T>>)	コレクションとして渡す場合
Merge(Observable<Observable<T>>)	ネストされた Observable をフラット化する場合

3. マーブルダイアグラム

複数のソースから到着した要素が、時系列順にそのまま下流へ流れます。

4. サンプルコード

using R3;

var subject1 = new Subject<string>();
var subject2 = new Subject<string>();

subject1.Merge(subject2)
    .Subscribe(x => Console.WriteLine(x));

subject1.OnNext("A");     // 出力: A
subject2.OnNext("1");     // 出力: 1
subject1.OnNext("B");     // 出力: B
subject2.OnNext("2");     // 出力: 2

ファクトリメソッド版 Merge の例:

using R3;

var s1 = new Subject<int>();
var s2 = new Subject<int>();
var s3 = new Subject<int>();

Observable.Merge(s1, s2, s3)
    .Subscribe(x => Console.WriteLine(x));

s2.OnNext(20);    // 出力: 20
s1.OnNext(10);    // 出力: 10
s3.OnNext(30);    // 出力: 30

5. 補足

Concat との違い

Concat は前の Observable が完了してから次の Observable を購読します。Merge はすべてのソースに同時に購読するため、要素はソース単位ではなく到着順に流れます。

Switch との違い

Switch は最新の内部 Observable のみを購読し、以前の購読を破棄します。Merge はすべての内部 Observable を同時に購読し続けます。

ファクトリメソッド版との違い

ファクトリメソッド版 Merge は、Observable.Merge(source1, source2, source3) のように複数ソースをまとめて指定する場合に使います。2 つのソースを読みやすく合流させたい場合は、このオペレーター版の source.Merge(second) が向いています。