Lifetime of 'Linqed' IObservable using Reactive Expressions

Question

This is potentially a very stupid question, but I'm not sure if I should be holding on to the IObservables generated by the Linq extension methods. Most of the examples seem to do so by default, but what would happen in a long running program. I have two examples to explain my question:

My first, and currently default implementation, involves subscribing to an IObservable provided by a pub\sub event aggregator. I'm comfortable with this because I know the observable will be around as long as the aggregator is.

protected virtual void SubscribeToEvents()
{
    _subscriptions.Dispose();
    _subscriptions = new CompositeDisposable();

    IDisposable sendSubscription = Events.GetStream<OutgoingByteMessage>()
        .Subscribe(msg => 
        {
            if (msg.Identifier.ChannelId == this.Id)
            {
                var queueItem = Tuple.Create(msg.Identifier.ConnectionId, msg.Data);
                _sendQueue.Enqueue(queueItem);
            }
        });            

    _subscriptions.Add(sendSubscription);                       
} 

However, if I use Linq to break down some of that logic, I can do this to get a subscription:

IDisposable altSendSubscription = Events.GetStream<OutgoingByteMessage>()
    .Where(msg => msg.Identifier.ChannelId == this.Id)
    .Select(msg => Tuple.Create(msg.Identifier.ConnectionId, msg.Data))
    .Subscribe(item =>
    {                    
        _sendQueue.Enqueue(item);                    
    });

My question is, on that second version, should I be keeping the product of that .Select method? This is happening in a method, afterall, so technically wouldn't the IObservable being returned there go out of scope as soon as SubscribeToEvents complete and be eligible for garbage collection? Or does the IDisposable object itself keep a reference to it, thereby making it safe to not keep a concrete reference to it somewhere.

After looking at the Rx source it looks like they use a series of forwarder classes to accomplish the Linq magic but I'm still a little paranoid. What do you guys think?

Answer 1

At the end of the day, all of the IObservable objects are being held via the list of subscribers on Events.GetStream (depending on your implementation of Events.GetStream, of course). Whether this goes out of scope depends on your implementation.

However, holding the IDisposable almost certainly keeps the subscription in scope as well, so even if your implementation was return Observable.Never<OutgoingByteMessage>, it would still exist because of the IDisposable.

You usually don't have to worry about active subscriptions going out of scope - the thing you should worry instead about is, subscription leaks. These occur when you subscribe to a global object (like something returned from an IoC container) and don't disconnect the subscription (and if the Observable never terminates).

This sounds like it'd not happen often, but it comes up more than you'd expect. For example:

MessageBus.Current.Listen<FooMessage>()
    .Where(x => x.Bar == "Baz")
    .Subscribe(x => Console.WriteLine(x));

If I put that in a constructor of some object that gets created all the time, I create dozens of subscriptions that never get cleaned up.

Answer 2

No, you do not need to worry. Think about it like this: as long as there is a source of events, the observable chain cannot be collectable since the act of subscribing links them all together. That is, each operator subscribes to the one above creating an observer it's parent references.

You don't even need to hold on to the subscription handle (altSendSubscription).

Why not use a memory profiler and look at the heap to convince yourself?