How can I create an Rx observable which stops publishing events when the last observer unsubscribes?

Question

I'll create an observable (through a variety of means) and return it to interested parties, but when they're done listening, I'd like to tear down the observable so it doesn't continue consuming resources. Another way to think of it as creating topics in a pub sub system. When no one is subscribed to a topic any more, you don't want to hold the topic and its filtering around anymore.

Answer 1

Rx already has an operator to suit your needs - well two actually - Publish & RefCount.

Here's how to use them:

IObservable xs = ...

var rxs = xs.Publish().RefCount();

var sub1 = rxs.Subscribe(x => { });
var sub2 = rxs.Subscribe(x => { });

//later
sub1.Dispose();

//later 
sub2.Dispose();

//The underlying subscription to `xs` is now disposed of.

Simple.

Answer 2

If I have understood your question you want to create the observable such that when all subscribers have disposed their subscription i.e there is no more subscriber, then you want to execute a clean up function which will stop the observable from production further values. If this is what you want then you can do something like below:

//Wrap a disposable
public class WrapDisposable : IDisposable
    {
        IDisposable disp;
        Action act;
        public WrapDisposable(IDisposable _disp, Action _act)
        {
            disp = _disp;
            act = _act;
        }
        void IDisposable.Dispose()
        {
            act();
            disp.Dispose();
        }
    }

    //Observable that we want to clean up after all subs are done
    public static IObservable<long> GenerateObs(out Action cleanup)
    {
        cleanup = () =>
        {
            Console.WriteLine("All subscribers are done. Do clean up");
        };
        return Observable.Interval(TimeSpan.FromSeconds(1));
    }
    //Wrap the observable
    public static IObservable<T> WrapToClean<T>(IObservable<T> obs, Action onAllDone)
    {
        int count = 0;
        return Observable.CreateWithDisposable<T>(ob =>
        {
            var disp = obs.Subscribe(ob);
            Interlocked.Increment(ref count);
            return new WrapDisposable(disp,() =>
            {
                if (Interlocked.Decrement(ref count) == 0)
                {
                    onAllDone();                                                
                }
            });
        });
    }

//Usage example:

Action cleanup;
var obs = GenerateObs(out cleanup);
var newObs = WrapToClean(obs, cleanup);
newObs.Take(6).Subscribe(Console.WriteLine);
newObs.Take(5).Subscribe(Console.WriteLine);