2

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);
}
4
  • What is the desirable behavior in case an observer is subscribed after the subj.OnNext(3); and before the subj.OnCompleted();? Should it receive one or two OnNext(1) notifications? Commented Jun 9, 2022 at 11:53
  • 1
    @TheodorZoulias, in this case the observer should receive one OnNext notification, immediately followed by OnCompleted notification. It's a one-off event cache.
    – noseratio
    Commented Jun 9, 2022 at 12:01
  • @noseratio - Please note that once an subject is complete it cannot emit more values. That is the contract in Rx. Your sample code can never have the output you expect. Commented Jul 3, 2022 at 5:17
  • 1
    @Enigmativity, FWIW, 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.
    – noseratio
    Commented Jul 3, 2022 at 8:46

3 Answers 3

1

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.

5
  • 1
    👍and I think it's safe to replace ReplaySubject<T> subject = new(1) with AsyncSubject<T> subject = new(), which has the "replay final item" behavior.
    – noseratio
    Commented Jul 2, 2022 at 5:53
  • 2
    Hmm, funny, I initially used just a single replay subject as well and completed on the first OnNext() call when I initially looked at this, then decided it would be safer to have a separate subject. Thinking about it again, I guess it wasn't necessary after all. Commented Jul 3, 2022 at 0:05
  • On a side note, what do you folks think about Subjects and the Rx contract? Genuinely interested. Is it safe to say that only IObservable endpoint of a subject is bound by it?
    – noseratio
    Commented Jul 4, 2022 at 3:05
  • 1
    @noseratio this question sounds a bit theoretical to me. You could consider opening a new question about it, although it might be closed as "opinion based". I think that the Rx in general tends to attract theoretical minds, who enjoy thinking about how things should be done in an ideal world. I consider myself to be more of a practical guy. If it works, it works. :-) Commented Jul 4, 2022 at 4:14
  • 1
    @TheodorZoulias, likewise! 🙂
    – noseratio
    Commented Jul 4, 2022 at 5:32
1

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);
}
5
  • Why not use a TCS as the base instead?
    – Aron
    Commented Jul 2, 2022 at 1:34
  • Well that's one implementation the OP explored already, but this can be implemented using RX constructs. Commented Jul 2, 2022 at 1:37
  • I meant wrapping a TCS instead of a Subject. That will definitely have the behavior of a TCS.
    – Aron
    Commented Jul 2, 2022 at 1:56
  • Tks! I've moved on with TCS (as in the q), but this is rather elegant. Probably could replace ReplaySubject of 1 item buffer with AsyncSubject?
    – noseratio
    Commented Jul 2, 2022 at 3:27
  • 1
    @noseratio Oh yes, I believe that would work. I've never used that one but it sounds like it would do the job nicely. Commented Jul 2, 2022 at 5:59
1

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.

11
  • 1
    @noseratio yep, your implementation is probably as good as it gets! The Lazy<T> might be redundant, but it shouldn't be harmful either. Commented Jun 9, 2022 at 17:31
  • 1
    Theodor, tks! I kinda hoped it can be composed with Rx operators but oh well...
    – noseratio
    Commented Jun 9, 2022 at 21:43
  • 1
    @noseratio I don't think that the ReplaySubject<T>(1) idea can work. If you push two messages before any observer has subscribed, most likely the second message will overwrite the first. Commented Jun 9, 2022 at 23:06
  • 1
    Yeah you're probably right. I also wary side effects in the pipleline, like Do or Finally.
    – noseratio
    Commented Jun 9, 2022 at 23:18
  • 1
    @noseratio regarding the allocations associated with the Lazy<T>, check out this demo. Comment the #define USING_LAZY line at the top, to see whether the allocations are increased or reduced. :-) Commented Jun 9, 2022 at 23:24

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.