Multicast

1. 概要

Multicast は、ユーザーが提供した ISubject<T> を介してソースをマルチキャストする ConnectableObservable を返すオペレーターです。使用する Subject の種類を自由に選択できるため、マルチキャストの動作を完全にカスタマイズできます。

多くの典型的な Subject 指定には専用オペレーターがあります。次の宣言は、それぞれ右側のオペレーターを使うことと同義です。

Multicast での宣言	同義のオペレーター
source.Multicast(new Subject<T>())	source.Publish()
source.Multicast(new BehaviorSubject<T>(initialValue))	source.Publish(initialValue)
source.Multicast(new ReplaySubject<T>())	source.Replay()
source.Multicast(new ReplaySubject<T>(bufferSize))	source.Replay(bufferSize)
source.Multicast(new ReplaySubject<T>(window))	source.Replay(window)
source.Multicast(new ReplaySubject<T>(window, timeProvider))	source.Replay(window, timeProvider)
source.Multicast(new ReplaySubject<T>(bufferSize, window))	source.Replay(bufferSize, window)
source.Multicast(new ReplaySubject<T>(bufferSize, window, timeProvider))	source.Replay(bufferSize, window, timeProvider)
source.Multicast(new ReplayFrameSubject<T>(window))	source.ReplayFrame(window)
source.Multicast(new ReplayFrameSubject<T>(window, frameProvider))	source.ReplayFrame(window, frameProvider)
source.Multicast(new ReplayFrameSubject<T>(bufferSize, window))	source.ReplayFrame(bufferSize, window)
source.Multicast(new ReplayFrameSubject<T>(bufferSize, window, frameProvider))	source.ReplayFrame(bufferSize, window, frameProvider)

2. シグネチャ

ISubject を指定

public static ConnectableObservable<T> Multicast<T>(
    this Observable<T> source,
    ISubject<T> subject)

指定した subject を介してソースの通知をマルチキャストする ConnectableObservable を返します。Connect() を呼び出すと、ソースが subject に購読され、subject を通じてすべての購読者に通知が配信されます。

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

Connect() 後、ソースの通知は指定した Subject を介してすべての購読者に配信されます。Subject の種類によって動作が異なります。

4. サンプルコード

// Subject を使ったマルチキャスト
var subject = new Subject<int>();
var connectable = observable.Multicast(subject);

connectable.Subscribe(x => Console.WriteLine($"購読者1: {x}"));
connectable.Subscribe(x => Console.WriteLine($"購読者2: {x}"));

var connection = connectable.Connect();

// ReplaySubject を使用（Replay() と同等）
var replaySubject = new ReplaySubject<int>(bufferSize: 5);
var replayConnectable = observable.Multicast(replaySubject);

replayConnectable.Connect();
// 後から購読しても過去の値を受け取る
replayConnectable.Subscribe(x => Console.WriteLine($"後から購読: {x}"));

// BehaviorSubject を使用（Publish(initialValue) と同等）
var behaviorSubject = new BehaviorSubject<string>("初期値");
var behaviorConnectable = observable.Multicast(behaviorSubject);
behaviorConnectable.Subscribe(x => Console.WriteLine($"値: {x}")); // すぐに "初期値" を受け取る

5. 補足

通常は Publish、Replay、ReplayFrame などの専用オペレーターを使う方が意図が明確です。Multicast をあえて使うのは、専用オペレーターで表現できない ISubject<T> を使いたい場合です。

例えば、独自実装の ISubject<T> を挟んで通知を監視・記録したい場合、Subject インスタンスの生成やライフサイクルを呼び出し側で管理したい場合、またはテスト用の Subject を明示的に差し替えたい場合に Multicast が有用です。単に Subject<T>、BehaviorSubject<T>、ReplaySubject<T>、ReplayFrameSubject<T> を使うだけであれば、対応する専用オペレーターを優先してください。