In Rx.NET, how do I make a Subject to resemble TaskCompletionSource behavior?

user_1768303 0
Viewed 104

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);
}
rx.net reactivex system.reactive .net c#
3 Answers

What you describe sounds quite similar to the WriteOnceBlock<T> from the TPL Dataflow library. The dataflow blocks have a convenient extension method AsObservable, so based on this idea an implementation would look like this:

public class WriteOnceSubject<T> : ISubject<T>
{
    private readonly WriteOnceBlock<T> _block = new WriteOnceBlock<T>(x => x);

    public void OnCompleted() => _block.Complete();
    public void OnError(Exception error) => ((ISourceBlock<T>)_block).Fault(error);
    public void OnNext(T value) => _block.Post(value);

    public IDisposable Subscribe(IObserver<T> observer)
        => _block.AsObservable().Subscribe(observer);
}

Unfortunately this idea doesn't work. The subscribers of the WriteOnceSubject<T> are getting only an OnCompleted() notification. No OnNext() is emitted. I just posted a bug report on GitHub about this issue.

Update: Here is Microsoft's feedback regarding the bug report, by Stephen Toub:

WriteOnceBlock only ever has a single value, which is consumable any number of times, and as such the block completes as soon as it's been given a value. AsObservable checks whether a source has completed and takes that as an indication that no more data will be coming. So if you subscribe the observer prior to data being passed to the WriteOnceBlock, the WriteOnceBlock will dutifully propagate that data to linked targets prior to completing and the observer will receive it, but if the observer is subscribed after the WriteOnceBlock has completed, it'll assume no data is coming, and it'll itself signal completion.

It's possible those checks could be removed from AsObservable, at some expense if the source has already completed, but at present WriteOnceBlock composability with AsObservable isn't perfect.

user_111785490

You could write it in terms of two subjects, one replay subject to emit the value if set, and another to control initialization.

public class PromiseSubject<T> : ISubject<T>
{
    private readonly Subject<T> initialize = new();
    private readonly ReplaySubject<T> subject = new(1);
    public PromiseSubject() => initialize.Subscribe(subject);
    
    public void OnCompleted() => initialize.OnCompleted();
    public void OnError(Exception error) => initialize.OnError(error);
    public void OnNext(T value)
    {
        initialize.OnNext(value);
        initialize.OnCompleted();
    }

    public IDisposable Subscribe(IObserver<T> observer) => subject.Subscribe(observer);
}
user_3902780

This is a simplified version of Jeff Mercado's answer. I think that the desirable behavior can be achieved simply by completing a ReplaySubject(bufferSize: 1) after the first OnNext.

Actually an AsyncSubject<T>, as pointed out by @noseratio in a comment, is even simpler, and also slightly more efficient because it stores its single value in a field instead of an array.

public class WriteOnceSubject<T> : ISubject<T>
{
    private readonly AsyncSubject<T> subject = new();

    public void OnNext(T value) { subject.OnNext(value); subject.OnCompleted(); }
    public void OnError(Exception error) => subject.OnError(error);
    public void OnCompleted() => subject.OnCompleted();

    public IDisposable Subscribe(IObserver<T> observer) => subject.Subscribe(observer);
}

So in this sequence of events:

writeOnceSubject.OnNext(1);
writeOnceSubject.OnNext(2);
writeOnceSubject.OnNext(3);
writeOnceSubject.OnCompleted();
writeOnceSubject.OnError(new Exception());

...all commands except the first will be no-ops. When the writeOnceSubject is subscribed later, it will emit the value 1 that is stored in its buffer, followed by an OnCompleted notification.

user_111785490
Related