Push duplicated code in Delay into a base class. (#663)

Rx.NET/Source/src/System.Reactive/Linq/Observable/Delay.cs

@@ -25,12 +25,25 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal abstract class _ : IdentitySink<TSource>
             {
-                public _(IObserver<TSource> observer)
+                protected IStopwatch _watch;
+                protected IScheduler _scheduler;
+
+                public _(TParent parent, IObserver<TSource> observer)
                     : base(observer)
                 {
+                    _scheduler = parent._scheduler;
+                }
+
+                public void Run(TParent parent)
+                {
+                    _watch = _scheduler.StartStopwatch();
+
+                    RunCore(parent);
+
+                    base.Run(parent._source);
                 }
 
-                public abstract void Run(TParent parent);
+                protected abstract void RunCore(TParent parent);
             }
 
             internal abstract class S : _
@@ -38,15 +51,11 @@ namespace System.Reactive.Linq.ObservableImpl
                 protected readonly object _gate = new object();
                 protected IDisposable _cancelable;
 
-                protected readonly IScheduler _scheduler;
-
                 public S(TParent parent, IObserver<TSource> observer)
-                    : base(observer)
+                    : base(parent, observer)
                 {
-                    _scheduler = parent._scheduler;
                 }
 
-                protected IStopwatch _watch;
                 protected TimeSpan _delay;
                 protected bool _ready;
                 protected bool _active;
@@ -58,15 +67,6 @@ namespace System.Reactive.Linq.ObservableImpl
                 private bool _hasFailed;
                 private Exception _exception;
 
-                public override void Run(TParent parent)
-                {
-                    _watch = _scheduler.StartStopwatch();
-
-                    RunCore(parent);
-
-                    base.Run(parent._source);
-                }
-
                 protected override void Dispose(bool disposing)
                 {
                     base.Dispose(disposing);
@@ -77,8 +77,6 @@ namespace System.Reactive.Linq.ObservableImpl
                     }
                 }
 
-                protected abstract void RunCore(TParent parent);
-
                 public override void OnNext(TSource value)
                 {
                     var shouldRun = false;
@@ -257,15 +255,11 @@ namespace System.Reactive.Linq.ObservableImpl
                 protected IDisposable _cancelable;
                 private readonly SemaphoreSlim _evt = new SemaphoreSlim(0);
 
-                private readonly IScheduler _scheduler;
-
                 public L(TParent parent, IObserver<TSource> observer)
-                    : base(observer)
+                    : base(parent, observer)
                 {
-                    _scheduler = parent._scheduler;
                 }
 
-                protected IStopwatch _watch;
                 protected TimeSpan _delay;
                 protected Queue<System.Reactive.TimeInterval<TSource>> _queue = new Queue<System.Reactive.TimeInterval<TSource>>();
 
@@ -275,15 +269,6 @@ namespace System.Reactive.Linq.ObservableImpl
                 private bool _hasFailed;
                 private Exception _exception;
 
-                public override void Run(TParent parent)
-                {
-                    _watch = _scheduler.StartStopwatch();
-
-                    RunCore(parent);
-
-                    base.Run(parent._source);
-                }
-
                 protected override void Dispose(bool disposing)
                 {
                     base.Dispose(disposing);
@@ -294,8 +279,6 @@ namespace System.Reactive.Linq.ObservableImpl
                     }
                 }
 
-                protected abstract void RunCore(TParent parent);
-
                 protected void ScheduleDrain()
                 {
                     _stop = new CancellationTokenSource();