Class ObservableAsync<T>

Namespace

ReactiveUI.Extensions.Async

Assembly

ReactiveUI.Extensions.dll

Represents an asynchronous observable sequence that allows observers to subscribe and receive notifications asynchronously.

public abstract class ObservableAsync<T> : IObservableAsync<T>

Type Parameters

T

The type of the elements produced by the observable sequence.

Inheritance

object

ObservableAsync<T>

Implements

IObservableAsync<T>

Derived

ConnectableObservableAsync<T>

GroupedAsyncObservable<TKey, TValue>

BaseReplayLatestSubjectAsync<T>

BaseStatelessReplayLastSubjectAsync<T>

BaseStatelessSubjectAsync<T>

BaseSubjectAsync<T>

Extension Methods

ObservableAsync.AggregateAsync<T, TAcc>(IObservableAsync<T>, TAcc, Func<TAcc, T, CancellationToken, ValueTask<TAcc>>, CancellationToken)

ObservableAsync.AggregateAsync<T, TAcc>(IObservableAsync<T>, TAcc, Func<TAcc, T, TAcc>, CancellationToken)

ObservableAsync.AggregateAsync<T, TAcc, TResult>(IObservableAsync<T>, TAcc, Func<TAcc, T, TAcc>, Func<TAcc, TResult>, CancellationToken)

ObservableAsync.AllAsync<T>(IObservableAsync<T>, Func<T, bool>, CancellationToken)

ObservableAsync.AnyAsync<T>(IObservableAsync<T>, Func<T, bool>?, CancellationToken)

ObservableAsync.AnyAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.Cast<T, TResult>(IObservableAsync<T>)

ObservableAsync.CatchAndIgnoreErrorResume<T>(IObservableAsync<T>, Func<Exception, IObservableAsync<T>>)

ObservableAsync.Catch<T>(IObservableAsync<T>, Func<Exception, IObservableAsync<T>>, Func<Exception, CancellationToken, ValueTask>?)

ObservableAsync.CombineLatest<T1, T2, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, Func<T1, T2, TResult>)

ObservableAsync.CombineLatest<T1, T2, T3, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, IObservableAsync<T3>, Func<T1, T2, T3, TResult>)

ObservableAsync.CombineLatest<T1, T2, T3, T4, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, IObservableAsync<T3>, IObservableAsync<T4>, Func<T1, T2, T3, T4, TResult>)

ObservableAsync.CombineLatest<T1, T2, T3, T4, T5, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, IObservableAsync<T3>, IObservableAsync<T4>, IObservableAsync<T5>, Func<T1, T2, T3, T4, T5, TResult>)

ObservableAsync.CombineLatest<T1, T2, T3, T4, T5, T6, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, IObservableAsync<T3>, IObservableAsync<T4>, IObservableAsync<T5>, IObservableAsync<T6>, Func<T1, T2, T3, T4, T5, T6, TResult>)

ObservableAsync.CombineLatest<T1, T2, T3, T4, T5, T6, T7, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, IObservableAsync<T3>, IObservableAsync<T4>, IObservableAsync<T5>, IObservableAsync<T6>, IObservableAsync<T7>, Func<T1, T2, T3, T4, T5, T6, T7, TResult>)

ObservableAsync.CombineLatest<T1, T2, T3, T4, T5, T6, T7, T8, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, IObservableAsync<T3>, IObservableAsync<T4>, IObservableAsync<T5>, IObservableAsync<T6>, IObservableAsync<T7>, IObservableAsync<T8>, Func<T1, T2, T3, T4, T5, T6, T7, T8, TResult>)

ObservableAsync.Concat<T>(IObservableAsync<T>, IObservableAsync<T>)

ObservableAsync.ContainsAsync<T>(IObservableAsync<T>, T, IEqualityComparer<T>?, CancellationToken)

ObservableAsync.ContainsAsync<T>(IObservableAsync<T>, T, CancellationToken)

ObservableAsync.CountAsync<T>(IObservableAsync<T>, Func<T, bool>?, CancellationToken)

ObservableAsync.CountAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.Delay<T>(IObservableAsync<T>, TimeSpan, TimeProvider?)

ObservableAsync.DistinctBy<T, TKey>(IObservableAsync<T>, Func<T, TKey>)

ObservableAsync.DistinctBy<T, TKey>(IObservableAsync<T>, Func<T, TKey>, IEqualityComparer<TKey>)

ObservableAsync.DistinctUntilChangedBy<T, TKey>(IObservableAsync<T>, Func<T, TKey>)

ObservableAsync.DistinctUntilChangedBy<T, TKey>(IObservableAsync<T>, Func<T, TKey>, IEqualityComparer<TKey>)

ObservableAsync.DistinctUntilChanged<T>(IObservableAsync<T>)

ObservableAsync.DistinctUntilChanged<T>(IObservableAsync<T>, IEqualityComparer<T>)

ObservableAsync.Distinct<T>(IObservableAsync<T>)

ObservableAsync.Distinct<T>(IObservableAsync<T>, IEqualityComparer<T>)

ObservableAsync.Do<T>(IObservableAsync<T>, Action<T>?, Action<Exception>?, Action<Result>?)

ObservableAsync.Do<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask>?, Func<Exception, CancellationToken, ValueTask>?, Func<Result, ValueTask>?)

ObservableAsync.FirstAsync<T>(IObservableAsync<T>, Func<T, bool>, CancellationToken)

ObservableAsync.FirstAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.FirstOrDefaultAsync<T>(IObservableAsync<T>, Func<T, bool>, T?, CancellationToken)

ObservableAsync.FirstOrDefaultAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.FirstOrDefaultAsync<T>(IObservableAsync<T>, T?, CancellationToken)

ObservableAsync.ForEachAsync<T>(IObservableAsync<T>, Action<T>, CancellationToken)

ObservableAsync.ForEachAsync<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask>, CancellationToken)

ObservableAsync.GroupBy<TKey, TValue>(IObservableAsync<TValue>, Func<TValue, TKey>)

ObservableAsync.GroupBy<TKey, TValue>(IObservableAsync<TValue>, Func<TValue, TKey>, Func<TKey, ISubjectAsync<TValue>>)

ObservableAsync.LastAsync<T>(IObservableAsync<T>, Func<T, bool>, CancellationToken)

ObservableAsync.LastAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.LastOrDefaultAsync<T>(IObservableAsync<T>, Func<T, bool>, T?, CancellationToken)

ObservableAsync.LastOrDefaultAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.LastOrDefaultAsync<T>(IObservableAsync<T>, T?, CancellationToken)

ObservableAsync.LongCountAsync<T>(IObservableAsync<T>, Func<T, bool>?, CancellationToken)

ObservableAsync.LongCountAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.Merge<T>(IObservableAsync<T>, IObservableAsync<T>)

ObservableAsync.Multicast<T>(IObservableAsync<T>, ISubjectAsync<T>)

ObservableAsync.ObserveOn<T>(IObservableAsync<T>, AsyncContext, bool)

ObservableAsync.ObserveOn<T>(IObservableAsync<T>, IScheduler, bool)

ObservableAsync.ObserveOn<T>(IObservableAsync<T>, SynchronizationContext, bool)

ObservableAsync.ObserveOn<T>(IObservableAsync<T>, TaskScheduler, bool)

ObservableAsync.OfType<T, TResult>(IObservableAsync<T>)

ObservableAsync.OnDispose<T>(IObservableAsync<T>, Action)

ObservableAsync.OnDispose<T>(IObservableAsync<T>, Func<ValueTask>)

ObservableAsync.OnErrorResumeAsFailure<T>(IObservableAsync<T>)

ObservableAsync.Prepend<T>(IObservableAsync<T>, IEnumerable<T>)

ObservableAsync.Prepend<T>(IObservableAsync<T>, T)

ObservableAsync.Publish<T>(IObservableAsync<T>)

ObservableAsync.Publish<T>(IObservableAsync<T>, SubjectCreationOptions)

ObservableAsync.Publish<T>(IObservableAsync<T>, T)

ObservableAsync.Publish<T>(IObservableAsync<T>, T, BehaviorSubjectCreationOptions)

ObservableAsync.ReplayLatestPublish<T>(IObservableAsync<T>)

ObservableAsync.ReplayLatestPublish<T>(IObservableAsync<T>, ReplayLatestSubjectCreationOptions)

ObservableAsync.Retry<T>(IObservableAsync<T>)

ObservableAsync.Retry<T>(IObservableAsync<T>, int)

ObservableAsync.Scan<T, TAcc>(IObservableAsync<T>, TAcc, Func<TAcc, T, CancellationToken, ValueTask<TAcc>>)

ObservableAsync.Scan<T, TAcc>(IObservableAsync<T>, TAcc, Func<TAcc, T, TAcc>)

ObservableAsync.SelectMany<T, TResult>(IObservableAsync<T>, Func<T, IObservableAsync<TResult>>)

ObservableAsync.SelectMany<T, TResult>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask<IObservableAsync<TResult>>>)

ObservableAsync.SelectMany<T, TCollection, TResult>(IObservableAsync<T>, Func<T, IObservableAsync<TCollection>>, Func<T, TCollection, TResult>)

ObservableAsync.Select<T, TDest>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask<TDest>>)

ObservableAsync.Select<T, TDest>(IObservableAsync<T>, Func<T, TDest>)

ObservableAsync.SingleAsync<T>(IObservableAsync<T>, Func<T, bool>, CancellationToken)

ObservableAsync.SingleAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.SingleOrDefaultAsync<T>(IObservableAsync<T>, Func<T, bool>, T?, CancellationToken)

ObservableAsync.SingleOrDefaultAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.SingleOrDefaultAsync<T>(IObservableAsync<T>, T?, CancellationToken)

ObservableAsync.SkipWhile<T>(IObservableAsync<T>, Func<T, bool>)

ObservableAsync.SkipWhile<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask<bool>>)

ObservableAsync.Skip<T>(IObservableAsync<T>, int)

ObservableAsync.StartWith<T>(IObservableAsync<T>, IEnumerable<T>)

ObservableAsync.StartWith<T>(IObservableAsync<T>, T)

ObservableAsync.StartWith<T>(IObservableAsync<T>, params T[])

ObservableAsync.StatelessPublish<T>(IObservableAsync<T>)

ObservableAsync.StatelessPublish<T>(IObservableAsync<T>, T)

ObservableAsync.StatelessReplayLatestPublish<T>(IObservableAsync<T>)

ObservableAsync.SubscribeAsync<T>(IObservableAsync<T>)

ObservableAsync.SubscribeAsync<T>(IObservableAsync<T>, Action<T>, Action<Exception>?, Action<Result>?, CancellationToken)

ObservableAsync.SubscribeAsync<T>(IObservableAsync<T>, Action<T>, CancellationToken)

ObservableAsync.SubscribeAsync<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask>)

ObservableAsync.SubscribeAsync<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask>, Func<Exception, CancellationToken, ValueTask>?, Func<Result, ValueTask>?, CancellationToken)

ObservableAsync.SubscribeAsync<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask>, CancellationToken)

ObservableAsync.TakeUntil<T>(IObservableAsync<T>, CompletionObservableDelegate, TakeUntilOptions?)

ObservableAsync.TakeUntil<T>(IObservableAsync<T>, Func<T, bool>)

ObservableAsync.TakeUntil<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask<bool>>)

ObservableAsync.TakeUntil<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.TakeUntil<T>(IObservableAsync<T>, Task, TakeUntilOptions?)

ObservableAsync.TakeUntil<T, TOther>(IObservableAsync<T>, IObservableAsync<TOther>, TakeUntilOptions?)

ObservableAsync.TakeWhile<T>(IObservableAsync<T>, Func<T, bool>)

ObservableAsync.TakeWhile<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask<bool>>)

ObservableAsync.Take<T>(IObservableAsync<T>, int)

ObservableAsync.Throttle<T>(IObservableAsync<T>, TimeSpan, TimeProvider?)

ObservableAsync.Timeout<T>(IObservableAsync<T>, TimeSpan, IObservableAsync<T>, TimeProvider?)

ObservableAsync.Timeout<T>(IObservableAsync<T>, TimeSpan, TimeProvider?)

ObservableAsync.ToAsyncEnumerable<T>(IObservableAsync<T>, Func<Channel<T>>, Func<Exception, CancellationToken, ValueTask>?)

ObservableAsync.ToDictionaryAsync<T, TKey>(IObservableAsync<T>, Func<T, TKey>, IEqualityComparer<TKey>?, CancellationToken)

ObservableAsync.ToDictionaryAsync<T, TKey, TValue>(IObservableAsync<T>, Func<T, TKey>, Func<T, TValue>, IEqualityComparer<TKey>?, CancellationToken)

ObservableAsync.ToListAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.WaitCompletionAsync<T>(IObservableAsync<T>, CancellationToken)

ObservableAsync.Where<T>(IObservableAsync<T>, Func<T, bool>)

ObservableAsync.Where<T>(IObservableAsync<T>, Func<T, CancellationToken, ValueTask<bool>>)

ObservableAsync.Yield<T>(IObservableAsync<T>)

ObservableAsync.Zip<T1, T2>(IObservableAsync<T1>, IObservableAsync<T2>)

ObservableAsync.Zip<T1, T2, TResult>(IObservableAsync<T1>, IObservableAsync<T2>, Func<T1, T2, TResult>)

ObservableBridgeExtensions.ToObservable<T>(IObservableAsync<T>)

ReactiveExtensions.Schedule<T>(T, DateTimeOffset, IScheduler)

ReactiveExtensions.Schedule<T>(T, DateTimeOffset, IScheduler, Action<T>)

ReactiveExtensions.Schedule<T>(T, IScheduler, Func<T, T>)

ReactiveExtensions.Schedule<T>(T, TimeSpan, IScheduler)

ReactiveExtensions.Schedule<T>(T, TimeSpan, IScheduler, Action<T>)

ReactiveExtensions.Schedule<T>(T, TimeSpan, IScheduler, Func<T, T>)

Remarks

Implementations of this class enable the observer pattern for asynchronous event streams, allowing observers to receive notifications as data becomes available. Observers should dispose the returned subscription to stop receiving notifications and release resources. This class is intended to be used as a base for custom asynchronous observable implementations.

Constructors

ObservableAsync()

protected ObservableAsync()

Methods

SubscribeAsync(IObserverAsync<T>, CancellationToken)

Asynchronously subscribes the specified asynchronous observer to receive notifications from the observable sequence.

public ValueTask<IAsyncDisposable> SubscribeAsync(IObserverAsync<T> observer, CancellationToken cancellationToken)

Parameters

observer IObserverAsync<T>

The asynchronous observer that will receive notifications. Cannot be null.

cancellationToken CancellationToken

A cancellation token that can be used to cancel the subscription operation.

Returns

ValueTask<IAsyncDisposable>

A task that represents the asynchronous operation. The result contains an IAsyncDisposable that can be disposed to unsubscribe the observer.

Remarks

The returned IAsyncDisposable should be disposed when the observer no longer wishes to receive notifications. Multiple calls to this method with the same observer will result in multiple independent subscriptions.

SubscribeAsyncCore(IObserverAsync<T>, CancellationToken)

Subscribes the specified asynchronous observer to receive notifications from the observable sequence.

protected abstract ValueTask<IAsyncDisposable> SubscribeAsyncCore(IObserverAsync<T> observer, CancellationToken cancellationToken)

Parameters

observer IObserverAsync<T>

The observer that will receive asynchronous notifications. Cannot be null.

cancellationToken CancellationToken

A cancellation token that can be used to cancel the subscription operation.

Returns

ValueTask<IAsyncDisposable>

A task that represents the asynchronous subscription operation. The result contains an object that can be disposed to unsubscribe the observer.

Remarks

The returned disposable should be disposed when the observer no longer wishes to receive notifications. Multiple calls to this method may result in multiple independent subscriptions.