In Rx.NET, how do I make a Subject
to resemble TaskCompletionSource.Task
behavior?
It needs to cache and reply the first event, even if completed. Neither AsyncSubject
nor ReplaySubject(bufferSize: 1)
would do that.
For example (let's call it PromiseSubject
):
//var subj = new ReplaySubject<int>(bufferSize: 1);
var subj = new PromiseSubject<int>();
subj.Subscribe(i => Console.WriteLine(i));
subj.OnNext(1);
subj.OnNext(2);
subj.OnNext(3);
subj.OnCompleted();
subj.Subscribe(i => Console.WriteLine(i));
Console.ReadLine();
Expected output:
1
1
I can possibly cook it up using TaskCompletionSource
, TaskObservableExtensions.ToObservable
and a custom SubjectBase
-derived subject implementation, but is there an elegant way of doing it using a composition of Rx operators?
Updated, my initial attempt via TaskCompletionSource
:
public class PromiseSubject<T> : ISubject<T>
{
private readonly TaskCompletionSource<(bool HasValue, T Value)> _tcs;
private readonly IObservable<T> _observable;
public PromiseSubject()
{
_tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
_observable = _tcs.Task.ToObservable()
.Where(r => r.HasValue).Select(r => r.Value!);
}
public void OnCompleted() =>
_tcs.TrySetResult((false, default!));
public void OnError(Exception error) =>
_tcs.TrySetException(error);
public void OnNext(T value) =>
_tcs.TrySetResult((true, value));
public IDisposable Subscribe(IObserver<T> observer) =>
_observable.Subscribe(observer);
}
subj.OnNext(3);
and before thesubj.OnCompleted();
? Should it receive one or twoOnNext(1)
notifications?OnNext
notification, immediately followed byOnCompleted
notification. It's a one-off event cache.AsyncSubject
emits after (and only after) its completion, which I believe is its documented and desired behavior. It keeps emitting for new subscribers, too: dotnetfiddle.net/0SbFpM.