Dispose of Observable Items as they are generated

Question

I have an IObservable that generates items that are disposable, and it will generate a potentially infinite number of them over its lifetime. Because of this, I want to dispose of the last item each time a new item is generated, so the Using operator will not work for this. Is there a different Rx.NET operator that can accomplish this function?

Reincarnate

Answer 1

If you have a IObservable<IDisposable> source then do this to automatically dispose of the previous value and to clean up when the sequence ends:

IObservable<IDisposable> query =
    Observable.Create<IDisposable>(o =>
    {
        var serial = new SerialDisposable();
        return new CompositeDisposable(
            source.Do(x => serial.Disposable = x).Subscribe(o),
            serial);
    })

Unmoved

It also seems to work fine if source was Subject<IDisposable>. Can I use Subject or do I have to convert to observable first? I have it as subject.Do(x => serial.Disposable = x).Subscribe(o), – Politburo

@XavierJohn - A Subject is an Observable and an Observer. That's what they are. So no need to convert. – Unmoved

Answer 2

Here is a DisposePrevious operator, based on a slightly modified version of Enigmativity's solution.

/// <summary>Disposes the previous element of an observable sequence. The last
/// element is disposed when the observable sequence completes.</summary>
public static IObservable<T> DisposePrevious<T>(this IObservable<T> source)
    where T : IDisposable
{
    return Observable.Using(() => new SerialDisposable(), serial =>
        source.Do(x => serial.Disposable = x));
}

The SerialDisposable class...

Represents a disposable resource whose underlying disposable resource can be replaced by another disposable resource, causing automatic disposal of the previous underlying disposable resource.

Astigmatic

Answer 3

I came across this answer by Shlomo and adapted it for my own purposes:

public class DisposerProperty<T> : IDisposable, IObservable<T> where T : IDisposable
{
    private IDisposable Subscription { get; }
    private IObservable<T> Source { get; }

    public T Value { get; private set; }

    public DisposerProperty(IObservable<T> source, T defaultValue = default)
    {
        Value = defaultValue;
        Source = source;
        Subscription = source.Subscribe(t =>
                                        {
                                            Value?.Dispose();
                                            Value = t;
                                        });
    }

    public void Dispose() => Subscription.Dispose();

    /// <inheritdoc />
    public IDisposable Subscribe(IObserver<T> observer) => Source.Subscribe(observer);
}

Now, whenever I want this functionality, I just use a DisposerProperty<T> instead of subscribing to the observable directly.

Reincarnate

This is a bad idea, There's a way to do this with existing operators and types. – Unmoved

@Unmoved Why is this bad? Like, in what way might this backfire? – Reincarnate

Implementing your own observable classes is hard to get right. I wouldn't use it unless you had some serious testing behind it. – Unmoved