Track the (number of) observers in an Observable?

c#, system.reactive

Solution

I would simply use RefCount / Publish. I always feel like if I'm implementing IObservable I'm working way too hard.

myColdObservable.Publish().RefCount();

This will make your observable stop pulsing after everyone has disconnected. Here's a sample:

var coldObservable = Observable
    .Interval(TimeSpan.FromSeconds(1))
    .ObserveOn(Scheduler.TaskPool)
    .Select(_ => DoSomething());

var refCountObs = coldObservable.Publish().RefCount();

CompositeDisposable d = new CompositeDisposable();
d.Add(refCountObs.Subscribe(n => Console.WriteLine("First got: " + n)));
d.Add(refCountObs.Subscribe(n => Console.WriteLine("Second got: " + n)));
d.Add(refCountObs.Subscribe(n => Console.WriteLine("Third got: " + n)));

//Wait a bit for work to happen
System.Threading.Thread.Sleep(10000);

//Everyone unsubscribes
d.Dispose();

//Observe that DoSomething is not called.
System.Threading.Thread.Sleep(3000);

This does not cover the case where you actually want to know the number of subscribers, but I think this fits with your requirements of stopping work if there are no subscribers.

Problem

I have an observable which represents a stream of stock prices. If there are no observers on my observable sequence I'd like to be able to disconnect from the remote server that is supplying the stream of prices, but I don't want to do that until every observer has called Dispose(). Then in a similar fashion, when the first person calls Subscribe I'd like to reconnect to the remote server. Is there a way to figure out how many observers have called subscribe on an observable? Or perhaps a way to know when observers are calling Subscribe or Dispose?

Original source