Class BaseStatelessReplayLastSubjectAsync<T>

Namespace

ReactiveUI.Extensions.Async.Subjects

Assembly

ReactiveUI.Extensions.dll

Provides a base class for stateless, asynchronous subjects that replay the last value to new subscribers and support resuming after errors or completion. Designed for scenarios where observers may join at any time and should receive the most recent value, if available.

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

Type Parameters

T

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

Inheritance

object

ObservableAsync<T>

BaseStatelessReplayLastSubjectAsync<T>

Implements

ISubjectAsync<T>

IObserverAsync<T>

IAsyncDisposable

IObservableAsync<T>

Derived

ConcurrentStatelessReplayLatestSubjectAsync<T>

SerialStatelessReplayLastSubjectAsync<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 implements asynchronous observer and observable patterns, allowing derived classes to define custom notification, error handling, and completion behaviors. It manages observer subscriptions and ensures that the most recent value (if any) is replayed to new subscribers. Thread safety is provided for concurrent access and notification. Typical use cases include event streaming, stateful data flows, or scenarios where late subscribers should receive the latest state.

Constructors

BaseStatelessReplayLastSubjectAsync(Optional<T>)

Provides a base class for stateless, asynchronous subjects that replay the last value to new subscribers and support resuming after errors or completion. Designed for scenarios where observers may join at any time and should receive the most recent value, if available.

protected BaseStatelessReplayLastSubjectAsync(Optional<T> startValue)

Parameters

startValue Optional<T>

The optional initial value to be used as the starting point for the subject. If provided, this value is immediately available to new subscribers until a new value is published.

Remarks

This abstract class implements asynchronous observer and observable patterns, allowing derived classes to define custom notification, error handling, and completion behaviors. It manages observer subscriptions and ensures that the most recent value (if any) is replayed to new subscribers. Thread safety is provided for concurrent access and notification. Typical use cases include event streaming, stateful data flows, or scenarios where late subscribers should receive the latest state.

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 observers that the asynchronous operation has completed and performs necessary cleanup.

public ValueTask OnCompletedAsync(Result result)

Parameters

result Result

The result information to be provided to observers upon completion.

Returns

ValueTask

A task that represents the asynchronous notification operation.

Remarks

After this method is called, the list of observers is cleared and the internal state is reset. Subsequent calls to this method will have no effect on previously completed observers.

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

Invoked to asynchronously notify all observers that the sequence has completed, providing the final result.

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

Parameters

observers IReadOnlyList<IObserverAsync<T>>

The collection of observers to be notified of the sequence completion. Cannot be null.

result Result

The result to provide to observers upon completion. Represents the outcome of the observed sequence.

Returns

ValueTask

A ValueTask that represents the asynchronous notification operation.

Remarks

This method is called when the observed sequence has finished processing. Implementations should ensure that all observers are notified according to the completion semantics of the sequence. This method is intended to be overridden in derived classes to customize completion behavior.

OnErrorResumeAsync(Exception, CancellationToken)

Handles an error by notifying all observers asynchronously and allows the operation to resume without propagating the exception.

public ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken)

Parameters

error Exception

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

cancellationToken CancellationToken

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

Returns

ValueTask

A task that represents the asynchronous error notification operation.

Remarks

This method ensures that all registered observers are notified of the specified error. The operation completes when all observers have been notified or the cancellation is requested.

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

Handles an error by resuming asynchronous observation for the specified observers.

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

Parameters

observers IReadOnlyList<IObserverAsync<T>>

A read-only list of observers to notify or resume after the error occurs. Cannot be null.

error Exception

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

cancellationToken CancellationToken

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

Returns

ValueTask

A ValueTask that represents the asynchronous error handling operation.

Remarks

Implementations should ensure that error handling is performed in a way that allows observers to continue receiving notifications or to recover from the error, as appropriate. This method is intended to be overridden to provide custom error recovery strategies in asynchronous observer scenarios.

OnNextAsync(T, CancellationToken)

Asynchronously notifies all registered 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 task that represents the asynchronous notification operation.

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

Asynchronously notifies the specified observers of a new 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 deliver 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

Derived classes should implement this method to define how notifications are delivered to observers. The method should honor the provided cancellation token and ensure that all observers in the list are notified according to the desired semantics.

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 task that represents the asynchronous operation. The result contains a disposable object that can be used to unsubscribe the observer.

Remarks

If the observable has a current value, it is immediately sent to the observer upon subscription. Disposing the returned object will unsubscribe the observer and may reset the observable's state if there are no remaining subscribers.