Merge

1. 概要

Merge は、複数の Observable ソースを 1 つに合流させるファクトリメソッドです。すべてのソースからの値が到着順にそのまま転送されます。すべてのソースが完了すると、結果の Observable も完了します。

2. シグネチャ

params 配列

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

可変長引数で複数のソースを指定します。

Observable.Merge(sourceA, sourceB, sourceC)

IEnumerable

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

ソースのコレクションを渡す場合に使用します。IEnumerable<Observable<T>> の拡張メソッドとしても呼び出せます。

var sources = new[] { sourceA, sourceB };
sources.Merge()

Observable<Observable<T>>（高階 Observable の平坦化）

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

Observable の Observable（高階 Observable）を平坦化します。外側のソースが発行する内側の Observable をすべて合流させます。これは オペレーター版 Merge の拡張メソッドです。

outerObservable.Merge()

オーバーロードの使い分け

オーバーロード	使う場面
Merge<T>(params Observable<T>[])	ソース数がコンパイル時に決まっている場合
Merge<T>(IEnumerable<Observable<T>>)	ソースを動的に生成・収集する場合
Merge<T>(Observable<Observable<T>>)	高階 Observable を平坦化する場合

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

Merge は複数のソースからの値を到着順にそのまま 1 つのストリームに合流させます。すべてのソースが完了すると結果も完了します。いずれかのソースがエラーを発行した場合、結果も即座にエラーとなります。

4. サンプルコード

using R3;

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

Observable.Merge(subject1, subject2)
    .Subscribe(x => Console.WriteLine($"値: {x}"));

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

subject1.OnCompleted();
subject2.OnNext("3");     // 出力: 値: 3（subject1 完了後も subject2 は継続）
subject2.OnCompleted();   // すべて完了 → 結果も完了

// IEnumerable 版
var sources = new[] { subject1.AsObservable(), subject2.AsObservable() };
sources.Merge()
    .Subscribe(x => Console.WriteLine($"値: {x}"));

5. 補足

関連 API

• Concat — 順番に連結（前のソースが完了してから次を購読）
• Race — 最初に値を発行したソースのみ購読
• CombineLatest — 各ソースの最新値を組み合わせる

オペレーター版

オペレーター版 Merge は、拡張メソッドとして source.Merge(second) の形式で使用できます。2 つの Observable を合流させる場合に便利です。この静的ファクトリメソッド版は、3 つ以上のソースや動的なソース集合をまとめて渡す場合に向いています。