Class RxExtensions

Namespace

RxSharp.Extras

Assembly

RxSharp.dll

Extension methods widening an error-only Observable<T> to another element type.

public static class RxExtensions

Inheritance

System.Object

RxExtensions

Methods

AssumeNeverEmits<TResult>(Observable<Unit>)

Widens an Observable<T> that only ever calls IObserver<T>.OnError (such as FromCancellationToken(CancellationToken, Func<Exception>?) or Timeout(TimeSpan, Func<Exception>?, IScheduler?)) so it type-checks anywhere an Observable<T> is expected - most commonly as one of the branches passed to RaceWith<T>(Observable<T>, params Observable<T>[]) alongside a source that actually produces TResult values. Mirrors how rxjs relies on TypeScript structurally accepting Observable<never> anywhere an Observable<T> is expected; C# has no bottom type, so this exists to fill that gap explicitly.

public static Observable<TResult> AssumeNeverEmits<TResult>(this Observable<Unit> source)

Parameters

source Observable<Unit>

An observable that only ever errors, never emits.

Returns

Observable<TResult>

An Observable<T> that mirrors source's errors and throws if it ever emits.

Type Parameters

TResult

The element type to widen to.

Remarks

This is only safe for a source that is guaranteed, by contract, to never call IObserver<T>.OnNext. If it ever does, the returned observable throws InvalidOperationException from that point on, converting a real value into a crash rather than propagating it - there is no compile-time check for this, unlike TypeScript's never. Do not apply this to a source that might legitimately emit.

FilterAsync<T>(Observable<T>, Func<T, Task<bool>>)

An operator supporting an async predicate, mirroring Puppeteer's own filterAsync helper (implemented, like theirs, via mergeMap). For each source value, predicate is awaited and the value is re-emitted only if it resolves to true; values for which it resolves to false are dropped. Because the underlying mergeMap subscribes to every predicate task concurrently as source values arrive, results can be emitted out of the original source order if an earlier value's predicate task takes longer to complete than a later value's.

public static Observable<T> FilterAsync<T>(this Observable<T> source, Func<T, Task<bool>> predicate)

Parameters

source Observable<T>

The source sequence to filter.

predicate Func<T, Task<bool>>

An async predicate evaluated for each source value.

Returns

Observable<T>

An observable that emits only the source values whose predicate resolved to true.

Type Parameters

T

The type of the source (and, when kept, result) values.

FromCancellationToken(CancellationToken, Func<Exception>?)

An observable that never emits and errors as soon as cancellationToken is cancelled. Mirrors Puppeteer's fromAbortSignal, completing the shape that helper fills in JS on top of rxjs. If the token is already cancelled at subscribe time, IObserver<T>.OnError fires synchronously during subscription rather than waiting for a cancellation callback.

public static Observable<Unit> FromCancellationToken(CancellationToken cancellationToken, Func<Exception>? causeFactory = null)

Parameters

cancellationToken CancellationToken

The token whose cancellation should be surfaced as an error.

causeFactory Func<Exception>?

Produces the exception passed to IObserver<T>.OnError when the token is cancelled. Defaults to a new OperationCanceledException.

Returns

Observable<Unit>

An observable that never calls IObserver<T>.OnNext and only ever errors.

FromEventBuffered<TEventArgs>(Action<EventHandler<TEventArgs>>, Action<EventHandler<TEventArgs>>, int)

The common case of FromEventBuffered<TDelegate, TEventArgs>(Action<TDelegate>, Action<TDelegate>, Func<Action<TEventArgs>, TDelegate>, int) for standard EventHandler<TEventArgs>-shaped .NET events.

public static BufferedEventSource<TEventArgs> FromEventBuffered<TEventArgs>(Action<EventHandler<TEventArgs>> addHandler, Action<EventHandler<TEventArgs>> removeHandler, int bufferSize = 1)

Parameters

addHandler Action<EventHandler<TEventArgs>>

Called immediately, with the handler to add to the event.

removeHandler Action<EventHandler<TEventArgs>>

Called on Dispose(), with the same handler, to remove it from the event.

bufferSize int

The maximum number of most-recent payloads kept for replay. Defaults to 1 - fine if the caller only cares about the single most recent payload, but a real risk if the caller filters for a specific match among possibly-several payloads that could arrive in the pre-subscribe gap this method exists to cover: with the default of 1, a matching payload can be silently evicted from the buffer by a later non-matching one before anyone subscribes to see it. Pass a larger value whenever more than one payload could plausibly arrive before subscription and only some of them are what the caller is looking for.

Returns

BufferedEventSource<TEventArgs>

A handle exposing the buffered payloads as an observable, and detaching the handler on disposal.

Type Parameters

TEventArgs

The type of the event's payload.

FromEventBuffered<TDelegate, TEventArgs>(Action<TDelegate>, Action<TDelegate>, Func<Action<TEventArgs>, TDelegate>, int)

Like FromEvent<TDelegate, TEventArgs>(Action<TDelegate>, Action<TDelegate>, Func<Action<TEventArgs>, TDelegate>), but attaches the underlying event handler immediately - when this method is called - rather than lazily at subscribe time, buffering up to bufferSize payloads that arrive before any subscriber attaches and replaying them to the first subscriber(s).

public static BufferedEventSource<TEventArgs> FromEventBuffered<TDelegate, TEventArgs>(Action<TDelegate> addHandler, Action<TDelegate> removeHandler, Func<Action<TEventArgs>, TDelegate> conversion, int bufferSize = 1)

Parameters

addHandler Action<TDelegate>

Called immediately, with the handler to add to the event.

removeHandler Action<TDelegate>

Called on Dispose(), with the same handler, to remove it from the event.

conversion Func<Action<TEventArgs>, TDelegate>

Converts an Action<TEventArgs> callback into the event's actual delegate shape.

bufferSize int

The maximum number of most-recent payloads kept for replay. Defaults to 1 - fine if the caller only cares about the single most recent payload, but a real risk if the caller filters for a specific match among possibly-several payloads that could arrive in the pre-subscribe gap this method exists to cover: with the default of 1, a matching payload can be silently evicted from the buffer by a later non-matching one before anyone subscribes to see it. Pass a larger value whenever more than one payload could plausibly arrive before subscription and only some of them are what the caller is looking for.

Returns

BufferedEventSource<TEventArgs>

A handle exposing the buffered payloads as an observable, and detaching the handler on disposal.

Type Parameters

TDelegate

The delegate type of the event handler.

TEventArgs

The type of the event's payload.

Remarks

Ordinary FromEvent<TDelegate, TEventArgs>(Action<TDelegate>, Action<TDelegate>, Func<Action<TEventArgs>, TDelegate>) is cold: nothing is attached to the underlying event until Subscribe is called, matching every other RxSharp source. This is deliberately not that. It exists for the narrow case where a handler must be attached before synchronously checking some existing state that might already satisfy what the caller is waiting for (e.g. checking a collection for an already-matching item) without losing an event that fires in the real gap between attaching the raw handler and actually subscribing to the returned observable. That gap is provably real on .NET, where event delivery can run on another thread - unlike single-threaded JS, where rxjs's own fromEmitterEvent has no equivalent problem, since nothing can fire between two synchronous statements. Most code should use the ordinary, cold FromEvent<TDelegate, TEventArgs>(Action<TDelegate>, Action<TDelegate>, Func<Action<TEventArgs>, TDelegate>) instead; reach for this only when that gap is a real, provable race, not by default.

RaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, CancellationToken)

Overload of RaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, Func<Exception>, CancellationToken) using the default cause factory (OperationCanceledException/TimeoutException).

public static Observable<T> RaceWithSignalAndTimer<T>(this Observable<T> source, TimeSpan timeout, CancellationToken cancellationToken)

Parameters

source Observable<T>

The source sequence to race.

timeout TimeSpan

The overall duration before giving up with a timeout error.

cancellationToken CancellationToken

A token that, when cancelled, aborts the wait immediately.

Returns

Observable<T>

An observable that mirrors source unless the timeout or cancellation fires first.

Type Parameters

T

The type of values produced by source.

RaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, Func<Exception>?, CancellationToken)

Races source against cancellation and a timeout, whichever fires first. The non-retrying half of RxExtensions.RetryAndRaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, Func<Exception>, TimeSpan?, CancellationToken)

• use this directly for a single wait (e.g. "wait for the next matching event") that doesn't need retrying, and reach for the retrying combinator when it does.

public static Observable<T> RaceWithSignalAndTimer<T>(this Observable<T> source, TimeSpan timeout, Func<Exception>? causeFactory, CancellationToken cancellationToken)

Parameters

source Observable<T>

The source sequence to race.

timeout TimeSpan

The overall duration before giving up with a timeout error. A zero or negative value disables the timeout.

causeFactory Func<Exception>?

Produces the exception used for both the cancellation and timeout branches. Defaults to OperationCanceledException for cancellation and TimeoutException for the timeout, via the defaults of FromCancellationToken(CancellationToken, Func<Exception>?) and Timeout(TimeSpan, Func<Exception>?, IScheduler?) respectively. Since one factory covers both branches, a caller needing to tell the two apart by exception type should pass null here (so each branch keeps its own distinct default type) and catch/rethrow as needed at the call site.

cancellationToken CancellationToken

A token that, when cancelled, aborts the wait immediately.

Returns

Observable<T>

An observable that mirrors source unless the timeout or cancellation fires first.

Type Parameters

T

The type of values produced by source.

RetryAndRaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, CancellationToken)

Overload of RetryAndRaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, Func<Exception>, TimeSpan?, CancellationToken) using the default retry delay (50 milliseconds) and default cause factory (OperationCanceledException/TimeoutException).

public static Observable<T> RetryAndRaceWithSignalAndTimer<T>(this Observable<T> source, TimeSpan timeout, CancellationToken cancellationToken)

Parameters

source Observable<T>

The source sequence to retry.

timeout TimeSpan

The overall duration before the action is abandoned with a timeout error.

cancellationToken CancellationToken

A token that, when cancelled, aborts the whole operation immediately.

Returns

Observable<T>

An observable that retries source until it succeeds, times out, or is cancelled.

Type Parameters

T

The type of values produced by source.

RetryAndRaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, Func<Exception>?, TimeSpan?, CancellationToken)

The combinator behind Puppeteer's Locator actions (click/fill/hover/wait): retry the source on error, racing the whole thing against cancellation and a timeout. Mirrors Puppeteer's retryAndRaceWithSignalAndTimer: pipe(retry({delay}), raceWith(fromAbortSignal(...), timeout(...))). This is what lets a Locator action keep re-attempting a flaky operation (e.g. "find and click an element that may not have rendered yet") while still giving up promptly, either because the caller cancelled it or because it took longer than timeout — whichever happens first. Since the cancellation and timeout branches are Observable<T> sources that only ever error (see FromCancellationToken(CancellationToken, Func<Exception>?) and Timeout(TimeSpan, Func<Exception>?, IScheduler?)), the only way this combinator produces a value is if the retried source itself emits one before either branch fires.

public static Observable<T> RetryAndRaceWithSignalAndTimer<T>(this Observable<T> source, TimeSpan timeout, Func<Exception>? causeFactory, TimeSpan? retryDelay, CancellationToken cancellationToken)

Parameters

source Observable<T>

The source sequence to retry (e.g. a single Locator attempt that may throw).

timeout TimeSpan

The overall duration before the action is abandoned with a timeout error. A zero or negative value disables the timeout.

causeFactory Func<Exception>?

Produces the exception used for both the cancellation and timeout branches. Defaults to OperationCanceledException for cancellation and TimeoutException for the timeout, via the defaults of FromCancellationToken(CancellationToken, Func<Exception>?) and Timeout(TimeSpan, Func<Exception>?, IScheduler?) respectively.

retryDelay TimeSpan?

The delay between retry attempts. Defaults to 50 milliseconds.

cancellationToken CancellationToken

A token that, when cancelled, aborts the whole operation immediately.

Returns

Observable<T>

An observable that retries source until it succeeds, times out, or is cancelled.

Type Parameters

T

The type of values produced by source.

Timeout(TimeSpan, Func<Exception>?, IScheduler?)

An observable that never emits and errors after delay elapses, or never errors at all if delay is zero or negative (in which case the returned observable is Never<T>() and completes/errors are both suppressed forever). Mirrors Puppeteer's own timeout() helper. Internally built from Timer(TimeSpan, IScheduler?), so the delay is scheduled on scheduler like any other scheduled operator.

public static Observable<Unit> Timeout(TimeSpan delay, Func<Exception>? causeFactory = null, IScheduler? scheduler = null)

Parameters

delay TimeSpan

The duration to wait before erroring. A zero or negative value disables the timeout entirely.

causeFactory Func<Exception>?

Produces the exception thrown once delay elapses. Defaults to a new TimeoutException.

scheduler IScheduler

The scheduler used to run the timer. Defaults to the scheduler Timer(TimeSpan, IScheduler?) itself defaults to.

Returns

Observable<Unit>

An observable that never calls IObserver<T>.OnNext and, unless delay is non-positive, errors once it elapses.