Class BaseStatelessSubjectAsync<T>

Namespace

ReactiveUI.Extensions.Async.Subjects

Assembly

ReactiveUI.Extensions.dll

Provides a base class for implementing asynchronous, stateless subjects that broadcast notifications to multiple observers. Supports asynchronous notification, error handling, and completion signaling for observers subscribing to the subject.

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

Type Parameters

T

The type of the elements processed and broadcast by the subject.

Inheritance

object

ObservableAsync<T>

BaseStatelessSubjectAsync<T>

Implements

ISubjectAsync<T>

IObserverAsync<T>

IAsyncDisposable

IObservableAsync<T>

Derived

ConcurrentStatelessSubjectAsync<T>

SerialStatelessSubjectAsync<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 abstract class enables the creation of custom asynchronous subjects that do not maintain internal state between notifications. It manages observer subscriptions and provides extensibility points for handling value notifications, error recovery, and completion events. Implementations should override the core notification methods to define specific broadcasting or error-handling behaviors. Thread safety and observer management are handled by the base class, allowing derived classes to focus on notification logic.

Constructors

BaseStatelessSubjectAsync()

protected BaseStatelessSubjectAsync()

Methods

DisposeAsync()

Asynchronously releases resources used by the current instance.

public ValueTask DisposeAsync()

Returns

ValueTask

A ValueTask that represents the asynchronous dispose operation.

Remarks

After calling this method, the instance should not be used. This method is intended to be called when the object is no longer needed, to ensure that all resources are properly released.

OnCompletedAsync(Result)

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

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.

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

The collection 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 according to the desired notification semantics.

OnErrorResumeAsync(Exception, CancellationToken)

Handles an error by resuming the asynchronous operation, allowing observers to continue receiving notifications despite the specified exception.

public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken)

Parameters

error Exception

The exception that caused the error condition. Cannot be null.

cancellationToken CancellationToken

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

Returns

ValueTask

A ValueTask that represents the asynchronous resume operation.

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 should resume processing after the error. Cannot be null.

error Exception

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

cancellationToken CancellationToken

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

Returns

ValueTask

A ValueTask that represents the asynchronous error recovery operation.

Remarks

Implementations should ensure that observers are notified or resumed appropriately based on the error. This method is intended to be called when an error occurs during asynchronous observation, allowing custom error handling strategies.

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.

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. Cannot be null.

cancellationToken CancellationToken

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

Returns

ValueTask<IAsyncDisposable>

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

Remarks

Disposing the returned IAsyncDisposable will remove the observer from the subscription list. The subscription is established immediately upon calling this method.