Class RxExtensions

Namespace

ReactiveExtensionsSharp.Extras

Assembly

ReactiveExtensionsSharp.dll

Extension methods providing an async-predicate flavor of Filter.

public static class RxExtensions

Inheritance

System.Object

RxExtensions

Methods

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 = null)

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?

See FromEventBuffered<TDelegate, TEventArgs>(Action<TDelegate>, Action<TDelegate>, Func<Action<TEventArgs>, TDelegate>, int?)'s parameter of the same name.

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 = null)

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 null, which defers entirely to ReplaySubject<T>'s own default (unbounded) - this wrapper has no buffering opinion of its own beyond what the primitive it's built on already does. A caller filtering for a specific match among payloads that could arrive in the pre-subscribe gap this method exists to cover needs more than 1: with too small a buffer, a matching payload can be silently evicted by a later non-matching one before anyone subscribes to see it. Pass an explicit bound only if this source stays attached for a long time without being subscribed to and unbounded growth is a real concern for that specific usage - the gap this method targets is normally microseconds, not something an unbounded buffer meaningfully grows during.

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 ReactiveExtensionsSharp 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 Task<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

Task<T>

A task that resolves to source's first value 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's first value against cancellation and a timeout, whichever fires first. The non-retrying half of 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. Races at the Task level rather than the Observable<T> level: cancellation and the timeout never produce a value of type T, only ever fault or never complete, and C# has no bottom type to make an error-only Observable<Unit> type-check as Observable<T> the way TypeScript's never lets rxjs do it - racing plain Tasks sidesteps that entirely.

public static Task<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. 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

Task<T>

A task that resolves to source's first value unless the timeout or cancellation fires first.

Type Parameters

T

The type of values produced by source.

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

Races source's first value against an already-existing signal task and a timeout, whichever fires first. Use this instead of the CancellationToken overload when the "give up" condition already exists as a task elsewhere (e.g. a task that faults when a session closes), rather than one this combinator needs to build.

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

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 thrown once timeout elapses. Defaults to a new TimeoutException.

signal Task

An already-existing task that, by contract, only ever faults or never completes.

Returns

Task<T>

A task that resolves to source's first value unless the timeout or signal fires first.

Type Parameters

T

The type of values produced by source.

Remarks

signal is expected, by contract, to only ever fault or never complete. If it completes without faulting, that is a contract violation in the caller and surfaces as an InvalidOperationException rather than silently returning a value.

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 Task<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

Task<T>

A task 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. The cancellation and timeout branches only ever fault or never complete (see RxExtensions.RaceWithSignalAndTimer<T>(Observable<T>, TimeSpan, Func<Exception>, CancellationToken)), so the only way this combinator produces a value is if the retried source itself produces one before either branch fires.

public static Task<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.

retryDelay TimeSpan?

The delay between retry attempts. Defaults to 50 milliseconds.

cancellationToken CancellationToken

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

Returns

Task<T>

A task 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.