gistfile1.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace UniRxEx
{
    public static partial class Observable
    {
        public static IObservable<T> RefCountEx<T>(this IConnectableObservable<T> source)
        {
            var connection =  source.Connect();
            var gate = new object();
            var refCount = 0;
            var flag = false;

            return Observable.Create<T>(observer =>
            {
                var subscription = source.Subscribe(observer);

                lock (gate)
                {
                    if (++refCount == 1 && flag)
                    {
                        connection = source.Connect();
                    }
                }

                return Disposable.Create(() =>
                {
                    subscription.Dispose();
                    lock (gate)
                    {
                        if (--refCount == 0)
                        {
                            connection.Dispose(); // connection isn't null.
                            flag = true;
                        }
                    }
                });
            });
        }
    }
}