Using Subject to decouple Observable subscription and initialisation

Question

I have an API which exposes an IObservable Status. But this status depends on an underlying observable source which has to be initialised via Init.

What I'd like to do is protect the users from having to do things in the right order: as it currently stands, if they try to subscribe to the Status before performing an Init, they get an exception because they source is not initialised.

So I had the genius idea of using a Subject to decouple the two: the external user subscribing to my Status is just subscribing to the Subject, then when they call Init, I subscribe to the underlying service using my Subject.

The idea in code

private ISubject<bool> _StatusSubject = new Subject<bool>();
public IObservable<bool> Status { get { return _StatusSubject; } }

public void Init() 
{
    _Connection = new Connection();
    Underlying.GetDeferredObservable(_Connection).Subscribe(_StatusSubject);
}

However, from tests on a dummy project, the problem is that the initialisation 'wakes up' my underlying Observable by subscribing the Subject to it, even if nobody has yet subscribed to the subject. That's something I'd like to avoid if possible, but I'm not sure how...

(I'm also mindful of the received wisdom that "the general rule is that if you're using a subject then you're doing something wrong")

Answer 1

It seems like the concept you are missing is how to know when someone starts listening and only init your underlying source. Usually you use Observable.Create or one of its siblings (Defer, Using, ...) to do this.

Here's how to do it without a Subject:

private IObservable<bool> _status = Observable.Defer(() =>
{
    _Connection = new Connection();
    return Underlying.GetDeferredObservable(_Connection);
};

public IObservable<bool> Status { get { return _status; } }

Defer will not call the init code until someone actually subscribes.

But this has a couple of potential issues:

1. Each observer will make a new connection
2. When the observer unsubscribes, the connection is not cleaned up.

The 2nd issue is easy to solve, so let's do that first. Let's assume your Connection is disposable, in which case you can just do:

private IObservable<bool> _status = Observable
    .Using(() => new Connection(),
           connection => Underlying.GetDeferredObservable(connection));

public IObservable<bool> Status { get { return _status; } }

With this iteration, whenever someone subscribes, a new Connection is created and passed to the 2nd lamba method to construct the observable. Whenever the observer unsubscribes, the Connection is Disposed. If Connection is not a IDisposable, then you can use Disposable.Create(Action) to create an IDisposable which will run whatever action you need to run to cleanup the connection.

You still have the problem that each observer creates a new connection. We can use Publish and RefCount to solve that problem:

private IObservable<bool> _status = Observable
    .Using(() => new Connection(),
           connection => Underlying.GetDeferredObservable(connection))
    .Publish()
    .RefCount();

public IObservable<bool> Status { get { return _status; } }

Now, when the first observer subscribes, the connection will get created and the underlying observable will be subscribed. Subsequent observers will share the connection and will pick up the current status. When the last observer unsubscribes, the connection will be disposed and everything shut down. If another observer subscribes after that, it all starts back up again.

Underneath the hood, Publish is actually using a Subject to share the single observable source. And RefCount is tracking how many observers are currently observing.

Answer 2

I might be oversimplifying here, but let me take a whack at using Subject as requested:

Your Thingy:

public class Thingy
{
    private BehaviorSubject<bool> _statusSubject = new BehaviorSubject<bool>(false);    
    public IObservable<bool> Status
    {
        get
        {
            return _statusSubject;
        }
    }

    public void Init()
    {
        var c = new object();
        new Underlying().GetDeferredObservable(c).Subscribe(_statusSubject);
    }
}

A faked out Underlying:

public class Underlying
{
    public IObservable<bool> GetDeferredObservable(object connection)
    {
        return Observable.DeferAsync<bool>(token => {
            return Task.Factory.StartNew(() => {
                Console.WriteLine("UNDERLYING ENGAGED");
                Thread.Sleep(1000);
                // Let's pretend there's some static on the line...
                return Observable.Return(true)
                    .Concat(Observable.Return(false))
                    .Concat(Observable.Return(true));
            }, token);
        });
    }
}

The harness:

void Main()
{
    var thingy = new Thingy();
    using(thingy.Status.Subscribe(stat => Console.WriteLine("Status:{0}", stat)))
    {
        Console.WriteLine("Waiting three seconds to Init...");
        Thread.Sleep(3000);
        thingy.Init();
        Console.ReadLine();
    }
}

The output:

Status:False
Waiting three seconds to Init...
UNDERLYING ENGAGED
Status:True
Status:False
Status:True

Answer 3

Hm, having played with this, I don't think that I can do it just with a Subject.

Not yet finished testing/trying, but here's what I've currently come up with which seems to work, but it doesn't protect me from the problems with Subject, as I'm still using one internally.

public class ObservableRouter<T> : IObservable<T>
{
    ISubject<T> _Subject = new Subject<T>();
    Dictionary<IObserver<T>, IDisposable> _ObserverSubscriptions 
                               = new Dictionary<IObserver<T>, IDisposable>();
    IObservable<T> _ObservableSource;
    IDisposable _SourceSubscription;

    //Note that this can happen before or after SetSource
    public IDisposable Subscribe(IObserver<T> observer)
    {
        _ObserverSubscriptions.Add(observer, _Subject.Subscribe(observer));
        IfReadySubscribeToSource();
        return Disposable.Create(() => UnsubscribeObserver(observer));
    }

    //Note that this can happen before or after Subscribe
    public void SetSource(IObservable<T> observable)
    {
        if(_ObserverSubscriptions.Count > 0 && _ObservableSource != null) 
                  throw new InvalidOperationException("Already routed!");
        _ObservableSource = observable;
        IfReadySubscribeToSource();
    }

    private void IfReadySubscribeToSource()
    {
        if(_SourceSubscription == null &&
           _ObservableSource != null && 
           _ObserverSubscriptions.Count > 0)
        {
            _SourceSubscription = _ObservableSource.Subscribe(_Subject);
        }
    }

    private void UnsubscribeObserver(IObserver<T> observer)
    {
        _ObserverSubscriptions[observer].Dispose();
        _ObserverSubscriptions.Remove(observer);
        if(_ObserverSubscriptions.Count == 0)
        {
            _SourceSubscription.Dispose();
            _SourceSubscription = null;
        }
    }
}