Subscribe

1. 概要

Subscribe は、Observable を購読して OnNext / OnErrorResume / OnCompleted の通知を受け取るための基本 API です。購読を開始すると IDisposable が返され、これを Dispose することで購読を解除できます。

2. 代表的な使い方

var subscription = observable.Subscribe(
    onNext: value => Console.WriteLine($"値: {value}"),
    onErrorResume: ex => Console.WriteLine($"非致命的エラー: {ex.Message}"),
    onCompleted: result => Console.WriteLine($"完了: {result}"));

// 不要になったら購読解除
subscription.Dispose();

OnNext だけを指定する簡易的な購読もよく使われます。

var subscription = observable.Subscribe(value =>
{
    Console.WriteLine(value);
});

3. Dispose と購読解除

Subscribe の戻り値は、その購読を表す IDisposable です。Dispose すると、その購読者は以降の通知を受け取らなくなります。

var subscription = source.Subscribe(x => Console.WriteLine(x));

subscription.Dispose();

Unity では、購読を GameObject や Component の寿命に連動させるために AddTo と組み合わせるのが一般的です。

button.OnClickAsObservable()
    .Subscribe(_ => Play())
    .AddTo(this);

4. 厳密な意味での Subscribe

R3 の Observable<T> クラスに実装されている、最も基本になる Subscribe は、Observer<T> を引数として受け取るメソッドです。

public abstract class Observable<T>
{
    public IDisposable Subscribe(Observer<T> observer);
}

つまり、厳密には Observable は「Observer<T> を受け取り、その Observer に OnNext / OnErrorResume / OnCompleted を通知するもの」として定義されています。ソース Observable や各オペレーターの内部実装も、この Observer<T> を受け取る購読モデルを基礎にしています。

この記事で主に解説している次のような Action を渡す形式の Subscribe は、ObservableSubscribeExtensions に定義された拡張メソッドです。

observable.Subscribe(
    onNext: value => Console.WriteLine(value),
    onErrorResume: ex => Console.WriteLine(ex),
    onCompleted: result => Console.WriteLine(result));

この拡張メソッドは、渡された Action から内部的に Observer<T> を作成し、本来の source.Subscribe(observer) を呼び出すためのラッパーです。通常の利用ではこの簡易形式を使えば十分ですが、R3 の実装や独自 Observable / Observer を理解する場合は、Observer<T> を受け取る Subscribe が本来の入口である、と捉えると全体像を掴みやすくなります。

5. SubscribeAwait / ForEachAsync との違い

API	戻り値	主な用途
Subscribe	IDisposable	通知を受け取り続け、必要に応じて購読解除する
SubscribeAwait	IDisposable	非同期の OnNext ハンドラで購読する
ForEachAsync	Task	シーケンス全体の完了を await する

Subscribe は購読を開始してすぐに IDisposable を返します。シーケンスの完了を await したい場合や、失敗完了を try-catch で扱いたい場合は ForEachAsync を使用します。