rawObserver

1. 概要

rawObserver は、Create、Defer、SubscribeOnSynchronize など一部の API が、下流の Observer<T> をそのまま使うか、observer.Wrap() した中継用の Observer<T> を使うかを切り替えるためのオプションです。

observer.Wrap() は、元の Observer<T> を包む新しい Observer<T> を作る内部用の拡張メソッドです。wrapper は OnNext、OnErrorResume、OnCompleted を元の Observer へ転送し、wrapper 自身が破棄されたときは元の Observer も破棄します。

public static Observer<T> Wrap<T>(this Observer<T> observer)
{
    return new WrappedObserver<T>(observer);
}

R3 の Observer<T> は、自分が接続されている購読解除用の参照を 1 つ持ちます。Observable<T>.Subscribe は、SubscribeCore が返した IDisposable をその Observer に紐づけ、Observer が破棄されたときにその購読も破棄できるようにします。

var subscription = SubscribeCore(observer);
observer.SourceSubscription.Disposable = subscription;

このため、Create や Defer の内部でさらに別の Observable を購読する場合、同じ Observer に外側の購読と内側の購読を両方持たせると、購読解除用の参照が衝突しやすくなります。

observer.Wrap() で新しい Observer を 1 つ挟むと、外側の Observable は元の Observer に、内側の Observable は wrapper に、それぞれ購読解除用の参照を持たせられます。rawObserver は、この wrapper を挟むかどうかを選ぶオプションです。

通常は false のまま使います。false では R3 が中継用の WrappedObserver<T> を作り、その wrapper を発行元や内側の Observable に渡します。これにより、外側の購読と内側の購読を別々の Observer インスタンスに紐づけられます。

true にすると wrapper を作らず、下流の Observer<T> を直接渡します。余分な allocation を避けられますが、購読チェーンや破棄順序を自分で保証する必要があります。一般的なアプリケーションコードでは指定しないでください。

2. シグネチャ

Create で使われる rawObserver

public static Observable<T> Create<T>(
    Func<Observer<T>, IDisposable> subscribe,
    bool rawObserver = false)

rawObserver: false では、subscribe に渡される Observer<T> は observer.Wrap() された wrapper です。rawObserver: true では、下流の Observer<T> がそのまま渡されます。

Observable.Create<int>(observer =>
{
    observer.OnNext(1);
    observer.OnCompleted();
    return Disposable.Empty;
})

Defer で使われる rawObserver

public static Observable<T> Defer<T>(
    Func<Observable<T>> observableFactory,
    bool rawObserver = false)

rawObserver: false では、ファクトリが返した Observable を wrapper 付きの Observer で購読します。rawObserver: true では、下流の Observer を直接その Observable へ渡します。

Observable.Defer(() => Observable.Return(DateTime.Now))

SubscribeOnSynchronize で使われる rawObserver

public static Observable<T> SubscribeOnSynchronize<T>(
    this Observable<T> source,
    object gate,
    bool rawObserver = false)

gate でロックして購読するときに、下流 Observer を wrapper 経由で渡すか直接渡すかを切り替えます。通常は false のままにします。

source.SubscribeOnSynchronize(gate)

3. true / false の違い

値	渡される Observer	特徴	使う場面
false	observer.Wrap()	外側の購読と内側の購読を別々の Observer に紐づける	通常はこちら
true	下流の Observer<T> そのもの	wrapper allocation を避ける	R3 内部実装や、購読チェーンを保証できる低レベル実装

R3 の実装では、概念的には次のように分岐しています。

return subscribe(rawObserver ? observer : observer.Wrap());

Observer<T> は OnCompleted 後の通知を無視するなどの基本的な状態管理を持っています。したがって、rawObserver: true は「すべての安全チェックを無効にする」設定ではありません。より正確には、observer.Wrap() による中継 Observer を省略する設定です。

4. 注意が必要な例

Create の中で別の Observable をそのまま購読する場合は、rawObserver: false が自然です。

using R3;

using var publisher = new Subject<int>();

var source = Observable.Create<int, Subject<int>>(
    publisher,
    static (observer, state) =>
    {
        return state.Subscribe(observer);
    });

この形では、state.Subscribe(observer) に渡される observer が wrapper になるため、外側の Create の購読は元の Observer に、内側の Subject への購読は wrapper に紐づきます。

rawObserver: true にすると、下流の Observer<T> を内側の Observable に直接渡します。同じ Observer に外側と内側の購読解除用参照を割り当てる形になりやすく、破棄管理の前提を崩す可能性があります。

5. 補足

rawObserver は、通常の値変換やイベント変換を便利にするためのオプションではありません。迷う場合は指定しないでください。

値の発行ロジックを自分で書く場合は Create、購読時に Observable を生成したい場合は Defer、購読処理をロックで同期したい場合は SubscribeOnSynchronize を参照してください。