Small fix in LazyRefCount, added another overload. This version is ho…

…pefully equivalent to my pull request to Rx.net (dotnet/reactive#133). Once this is merged, this implementation will be removed.
ExRam · Sep 14, 2015 · d37d029 · d37d029
1 parent fa68bd3
commit d37d029
Showing 1 changed file with 21 additions and 11 deletions.
diff --git a/ExRam.Extensions/System/Reactive/Linq/ObservableExtensions (LazyRefCount).cs b/ExRam.Extensions/System/Reactive/Linq/ObservableExtensions (LazyRefCount).cs
@@ -13,6 +13,13 @@ namespace System.Reactive.Linq
 {
     public static partial class ObservableExtensions
     {
+        public static IObservable<T> LazyRefCount<T>(this IConnectableObservable<T> source, TimeSpan delay)
+        {
+            Contract.Requires(source != null);
+
+            return source.LazyRefCount(delay, Scheduler.Default);
+        }
+
         public static IObservable<T> LazyRefCount<T>(this IConnectableObservable<T> source, TimeSpan delay, IScheduler scheduler)
         {
             Contract.Requires(source != null);
@@ -25,32 +32,35 @@ public static IObservable<T> LazyRefCount<T>(this IConnectableObservable<T> sour
                 .Create<T>(
                     () =>
                     {
-                        var schedulerSubscription = new SingleAssignmentDisposable();
-
                         lock (syncRoot)
                         {
                             if (currentConnection == null)
                                 currentConnection = source.Connect();
 
-                            serial.Disposable = schedulerSubscription;
+                            serial.Disposable = new SingleAssignmentDisposable();
                         }
 
                         return Disposable
                             .Create(() =>
                             {
-                                schedulerSubscription.Disposable = scheduler.Schedule(schedulerSubscription, delay, (self, state) =>
+                                lock (syncRoot)
                                 {
-                                    lock (syncRoot)
+                                    var cancelable = (SingleAssignmentDisposable)serial.Disposable;
+
+                                    cancelable.Disposable = scheduler.Schedule(cancelable, delay, (self, state) =>
                                     {
-                                        if (object.ReferenceEquals(serial.Disposable, state))
+                                        lock (syncRoot)
                                         {
-                                            currentConnection.Dispose();
-                                            currentConnection = null;
+                                            if (object.ReferenceEquals(serial.Disposable, state))
+                                            {
+                                                currentConnection.Dispose();
+                                                currentConnection = null;
+                                            }
                                         }
-                                    }
 
-                                    return Disposable.Empty;
-                                });
+                                        return Disposable.Empty;
+                                    });
+                                }
                             });
                     },
                     source.Subscribe)