Class BaseSubjectAsync<T>

Namespace

ReactiveUI.Extensions.Async.Subjects

Assembly

ReactiveUI.Extensions.dll

Provides a base class for asynchronous subjects that support both publishing values to observers and receiving values asynchronously.

public abstract class BaseSubjectAsync<T> : ObservableAsync<T>, ISubjectAsync<T>, IObserverAsync<T>, IAsyncDisposable, IObservableAsync<T>

Type Parameters

T

The type of elements processed by the subject and observed by subscribers.

Inheritance

object

ObservableAsync<T>

BaseSubjectAsync<T>

Implements

ISubjectAsync<T>

IObserverAsync<T>

IAsyncDisposable

IObservableAsync<T>

Derived

ConcurrentSubjectAsync<T>

SerialSubjectAsync<T>

Inherited Members

ObservableAsync<T>.SubscribeAsync(IObserverAsync<T>, CancellationToken)

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

ObservableAsync.Wrap<T>(IObserverAsync<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>)

SubjectMixins.AsObserverAsync<T>(ISubjectAsync<T>)

SubjectMixins.MapValues<T>(ISubjectAsync<T>, Func<IObservableAsync<T>, IObservableAsync<T>>)

Remarks

This class enables the implementation of asynchronous subjects that can broadcast values, errors, and completion notifications to multiple observers. It manages observer registration, notification, and completion in a thread-safe manner. Derived classes should override the core notification methods to customize how observers are notified asynchronously. The subject supports asynchronous subscription and notification patterns, making it suitable for reactive and event-driven programming scenarios.

Constructors

BaseSubjectAsync()

protected BaseSubjectAsync()

Methods

DisposeAsync()

Asynchronously releases the unmanaged resources used by the object.

public ValueTask DisposeAsync()

Returns

ValueTask

A ValueTask that represents the asynchronous dispose operation.

OnCompletedAsync(Result)

Notifies all registered observers that the asynchronous operation has completed and provides the final result.

public ValueTask OnCompletedAsync(Result result)

Parameters

result Result

The result to deliver to observers upon completion. Cannot be null.

Returns

ValueTask

A ValueTask that represents the asynchronous notification operation. The task completes when all observers have been notified.

Remarks

If the operation has already completed, this method returns immediately without notifying observers again. This method is thread-safe and ensures that observers are notified only once.

OnCompletedAsyncCore(IReadOnlyList<IObserverAsync<T>>, Result)

Invoked to asynchronously notify all observers of the completion event with the specified result.

protected abstract ValueTask OnCompletedAsyncCore(IReadOnlyList<IObserverAsync<T>> observers, Result result)

Parameters

observers IReadOnlyList<IObserverAsync<T>>

A read-only list of observers to be notified. Cannot be null.

result Result

The result to provide to each observer upon completion.

Returns

ValueTask

A ValueTask that represents the asynchronous notification operation.

Remarks

Implementations should ensure that all observers are notified, and handle any exceptions that may occur during notification according to the desired error-handling policy.

OnErrorResumeAsync(Exception, CancellationToken)

Notifies all observers of an error and allows asynchronous error handling to resume observation.

public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken)

Parameters

error Exception

The exception that occurred and will be sent to observers. Cannot be null.

cancellationToken CancellationToken

A cancellation token that can be used to cancel the asynchronous error notification operation.

Returns

ValueTask

A ValueTask that represents the asynchronous operation of notifying observers of the error.

Remarks

If the sequence has already completed or an error has previously been signaled, this method has no effect.

OnErrorResumeAsyncCore(IReadOnlyList<IObserverAsync<T>>, Exception, CancellationToken)

Handles error recovery for the specified observers by resuming asynchronous processing after an error occurs.

protected abstract ValueTask OnErrorResumeAsyncCore(IReadOnlyList<IObserverAsync<T>> observers, Exception error, CancellationToken cancellationToken)

Parameters

observers IReadOnlyList<IObserverAsync<T>>

A read-only list of observers that are to be notified or resumed following the error. Cannot be null.

error Exception

The exception that triggered the error handling logic. Cannot be null.

cancellationToken CancellationToken

A cancellation token that can be used to cancel the asynchronous error recovery operation.

Returns

ValueTask

A ValueTask that represents the asynchronous error recovery operation.

Remarks

Override this method to implement custom error recovery strategies for observers. The method is called when an error occurs during asynchronous processing, allowing derived classes to determine how to resume or notify observers. If the operation is canceled via the provided cancellation token, the returned task should reflect the cancellation.

OnNextAsync(T, CancellationToken)

Asynchronously notifies all subscribed observers of a new value.

public ValueTask OnNextAsync(T value, CancellationToken cancellationToken)

Parameters

value T

The value to send to each observer.

cancellationToken CancellationToken

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

Returns

ValueTask

A ValueTask that represents the asynchronous notification operation.

Remarks

If the sequence has already completed, this method does not notify observers.

OnNextAsyncCore(IReadOnlyList<IObserverAsync<T>>, T, CancellationToken)

Asynchronously notifies the specified observers with the provided value.

protected abstract ValueTask OnNextAsyncCore(IReadOnlyList<IObserverAsync<T>> observers, T value, CancellationToken cancellationToken)

Parameters

observers IReadOnlyList<IObserverAsync<T>>

A read-only list of observers to be notified. Cannot be null.

value T

The value to send to each observer.

cancellationToken CancellationToken

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

Returns

ValueTask

A ValueTask that represents the asynchronous notification operation.

SubscribeAsyncCore(IObserverAsync<T>, CancellationToken)

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

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

Parameters

observer IObserverAsync<T>

The asynchronous observer that will receive notifications from the observable sequence.

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 a disposable object that can be used to unsubscribe the observer.

Remarks

If the observable sequence has already completed, the observer receives the completion notification immediately and the returned disposable is a no-op. Otherwise, the observer is added to the list of active observers and will receive future notifications until unsubscribed or the sequence completes.