Class OperatorHelper

Namespace

RxSharp

Assembly

RxSharp.dll

The building block every operator is written against. Mirrors rxjs's operate() helper.

public static class OperatorHelper

Inheritance

System.Object

OperatorHelper

Methods

Operate<TSource, TResult>(Observable<TSource>, Func<Observable<TSource>, Subscriber<TResult>, IDisposable?>)

Creates a new Observable<T> whose subscribe logic is init, given the upstream source and the downstream Subscriber<T> to push notifications to. This is the low-level primitive every built-in operator is implemented on top of.

public static Observable<TResult> Operate<TSource, TResult>(this Observable<TSource> source, Func<Observable<TSource>, Subscriber<TResult>, IDisposable?> init)

Parameters

source Observable<TSource>

The upstream source observable.

init Func<Observable<TSource>, Subscriber<TResult>, IDisposable?>

Called at subscribe time with the source and the downstream subscriber; returns the teardown logic for the subscription, if any.

Returns

Observable<TResult>

The new, operator-defined observable.

Type Parameters

TSource

The element type of the source observable.

TResult

The element type of the resulting observable.

Pipe<TSource, TResult>(Observable<TSource>, OperatorFunction<TSource, TResult>)

Applies an OperatorFunction<TSource, TResult> to source. Equivalent to calling op directly; exists to let operators be composed in a left-to-right, method-chained style.

public static Observable<TResult> Pipe<TSource, TResult>(this Observable<TSource> source, OperatorFunction<TSource, TResult> op)

Parameters

source Observable<TSource>

The source observable.

op OperatorFunction<TSource, TResult>

The operator function to apply.

Returns

Observable<TResult>

The result of applying op to source.

Type Parameters

TSource

The element type of the source observable.

TResult

The element type of the resulting observable.

SubscribeChild<TSource, TResult>(Observable<TSource>, Subscriber<TResult>, Action<TSource>?, Action<Exception>?, Action?)

Subscribes to source with a fresh inner Subscriber<T>, registered as a child of downstream before the subscribe call runs. This is what lets a downstream disposal (an early-completing operator further down the chain, or an external unsubscribe) cascade up and stop source immediately — including mid-loop for a fully-synchronous, self-checking source — instead of only after the whole synchronous call stack has already unwound. See CLAUDE.md's Learnings for the full story of the bug this fixes.

public static IDisposable? SubscribeChild<TSource, TResult>(this Observable<TSource> source, Subscriber<TResult> downstream, Action<TSource>? onNext = null, Action<Exception>? onError = null, Action? onComplete = null)

Parameters

source Observable<TSource>

The source observable to subscribe to.

downstream Subscriber<TResult>

The downstream subscriber the new inner subscriber is registered as a child of.

onNext Action<TSource>?

Called for each value from source.

onError Action<Exception>?

Called if source errors.

onComplete Action?

Called when source completes.

Returns

IDisposable?

null, always — see the remarks above for why.

Type Parameters

TSource

The element type of the source observable.

TResult

The element type of the downstream subscriber (usually unrelated to the callback signatures, just needed to type downstream).

Remarks

Only use this for an operator with exactly ONE inner subscription for its whole lifetime. An operator that creates a new inner subscription per value or per cycle (MergeMap, SwitchMap, Debounce, Window*, ...) needs to also call Remove(IDisposable) once each inner subscription naturally ends, or the downstream subscriber's finalizer list grows unboundedly — this helper does not do that for you.

Deliberately returns null, not the inner subscription: the inner subscriber is already registered as a child of downstream above, so if a caller did return src.SubscribeChild(...) as the tail expression of an Operate<TSource, TResult>(Observable<TSource>, Func<Observable<TSource>, Subscriber<TResult>, IDisposable?>) callback and this method returned the inner subscriber, Subscribe(IObserver<T>) would add that same object to downstream a second time (it adds whatever the callback returns as a teardown). Harmless in practice (disposing a Subscription twice is a no-op) but a wasted, permanent duplicate entry in the finalizer list for the operator's whole lifetime — returning null here is what keeps that list to exactly one entry per inner subscription.