Use Schedule calls with state (#558)

akarnokd · danielcweber · commit 196b1726a8ae · 2018-06-01T22:44:20.000+02:00
diff --git a/Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.cs b/Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.cs
@@ -56,9 +56,11 @@ protected override IDisposable SubscribeCore(IObserver<TSource> observer)
                 var d = new SerialDisposable();
                 d.Disposable = m;
 
-                m.Disposable = scheduler.Schedule(() =>
+                m.Disposable = scheduler.Schedule((source, observer, d),
+                (scheduler, state) =>
                 {
-                    d.Disposable = new ScheduledDisposable(scheduler, source.SubscribeSafe(observer));
+                    state.d.Disposable = new ScheduledDisposable(scheduler, state.source.SubscribeSafe(state.observer));
+                    return Disposable.Empty;
                 });
 
                 return d;
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/Delay.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/Delay.cs
@@ -434,7 +434,7 @@ private void DrainQueue(ICancelable cancel)
                         if (shouldWait)
                         {
                             var timer = new ManualResetEventSlim();
-                            _scheduler.Schedule(waitTime, () => { timer.Set(); });
+                            _scheduler.Schedule(timer, waitTime, (_, slimTimer) => { slimTimer.Set(); return Disposable.Empty; });
 
                             try
                             {
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/FromEvent.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/FromEvent.cs
@@ -284,12 +284,13 @@ public IDisposable Connect(IObserver<TArgs> observer)
                         {
                             if (--_count == 0)
                             {
-                                _parent._scheduler.Schedule(_removeHandler.Dispose);
+                                _parent._scheduler.ScheduleAction(_removeHandler, handler => handler.Dispose());
                                 _parent._session = null;
                             }
                         }
                     });
                 }
+
             }
 
             private void Initialize()
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/Return.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/Return.cs
@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information. 
 
 using System.Reactive.Concurrency;
+using System.Reactive.Disposables;
 
 namespace System.Reactive.Linq.ObservableImpl
 {
@@ -33,7 +34,7 @@ public _(TResult value, IObserver<TResult> observer, IDisposable cancel)
 
             public IDisposable Run(IScheduler scheduler)
             {
-                return scheduler.Schedule(Invoke);
+                return scheduler.ScheduleAction(this, @this => @this.Invoke());
             }
 
             private void Invoke()
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/Skip.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/Skip.cs
@@ -101,14 +101,15 @@ public _(IObserver<TSource> observer, IDisposable cancel)
 
                 public IDisposable Run(Time parent)
                 {
-                    var t = parent._scheduler.Schedule(parent._duration, Tick);
+                    var t = parent._scheduler.Schedule(this, parent._duration, (_, state) => state.Tick());
                     var d = parent._source.SubscribeSafe(this);
                     return StableCompositeDisposable.Create(t, d);
                 }
 
-                private void Tick()
+                private IDisposable Tick()
                 {
                     _open = true;
+                    return Disposable.Empty;
                 }
 
                 public override void OnNext(TSource value)
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/SkipUntil.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/SkipUntil.cs
@@ -205,14 +205,15 @@ public _(IObserver<TSource> observer, IDisposable cancel)
 
             public IDisposable Run(SkipUntil<TSource> parent)
             {
-                var t = parent._scheduler.Schedule(parent._startTime, Tick);
+                var t = parent._scheduler.Schedule(this, parent._startTime, (_, state) => state.Tick());
                 var d = parent._source.SubscribeSafe(this);
                 return StableCompositeDisposable.Create(t, d);
             }
 
-            private void Tick()
+            private IDisposable Tick()
             {
                 _open = true;
+                return Disposable.Empty;
             }
 
             public override void OnNext(TSource value)
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/Take.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/Take.cs
@@ -112,17 +112,18 @@ public IDisposable Run(Time parent)
                 {
                     _gate = new object();
 
-                    var t = parent._scheduler.Schedule(parent._duration, Tick);
+                    var t = parent._scheduler.Schedule(this, parent._duration, (_, state) => state.Tick());
                     var d = parent._source.SubscribeSafe(this);
                     return StableCompositeDisposable.Create(t, d);
                 }
 
-                private void Tick()
+                private IDisposable Tick()
                 {
                     lock (_gate)
                     {
                         ForwardOnCompleted();
                     }
+                    return Disposable.Empty;
                 }
 
                 public override void OnNext(TSource value)
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/TakeUntil.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/TakeUntil.cs
@@ -180,17 +180,18 @@ public _(IObserver<TSource> observer, IDisposable cancel)
 
             public IDisposable Run(TakeUntil<TSource> parent)
             {
-                var t = parent._scheduler.Schedule(parent._endTime, Tick);
+                var t = parent._scheduler.Schedule(this, parent._endTime, (_, state) => state.Tick());
                 var d = parent._source.SubscribeSafe(this);
                 return StableCompositeDisposable.Create(t, d);
             }
 
-            private void Tick()
+            private IDisposable Tick()
             {
                 lock (_gate)
                 {
                     ForwardOnCompleted();
                 }
+                return Disposable.Empty;
             }
 
             public override void OnNext(TSource value)
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/Throw.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/Throw.cs
@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information. 
 
 using System.Reactive.Concurrency;
+using System.Reactive.Disposables;
 
 namespace System.Reactive.Linq.ObservableImpl
 {
@@ -33,13 +34,9 @@ public _(Exception exception, IObserver<TResult> observer, IDisposable cancel)
 
             public IDisposable Run(IScheduler scheduler)
             {
-                return scheduler.Schedule(Invoke);
+                return scheduler.ScheduleAction(this, @this => @this.ForwardOnError(@this._exception));
             }
 
-            private void Invoke()
-            {
-                ForwardOnError(_exception);
-            }
         }
     }
 }
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/Timeout.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/Timeout.cs
@@ -173,14 +173,14 @@ public IDisposable Run(Absolute parent)
 
                     _switched = false;
 
-                    var timer = parent._scheduler.Schedule(parent._dueTime, Timeout);
+                    var timer = parent._scheduler.Schedule(this, parent._dueTime, (_, state) => state.Timeout());
 
                     original.Disposable = parent._source.SubscribeSafe(this);
 
                     return StableCompositeDisposable.Create(_subscription, timer);
                 }
 
-                private void Timeout()
+                private IDisposable Timeout()
                 {
                     var timerWins = false;
 
@@ -192,6 +192,8 @@ private void Timeout()
 
                     if (timerWins)
                         _subscription.Disposable = _other.SubscribeSafe(GetForwarder());
+
+                    return Disposable.Empty;
                 }
 
                 public override void OnNext(TSource value)
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/Timer.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/Timer.cs
@@ -58,18 +58,19 @@ public _(IObserver<long> observer, IDisposable cancel)
 
                 public IDisposable Run(Single parent, DateTimeOffset dueTime)
                 {
-                    return parent._scheduler.Schedule(dueTime, Invoke);
+                    return parent._scheduler.Schedule(this, dueTime, (_, state) => state.Invoke());
                 }
 
                 public IDisposable Run(Single parent, TimeSpan dueTime)
                 {
-                    return parent._scheduler.Schedule(dueTime, Invoke);
+                    return parent._scheduler.Schedule(this, dueTime, (_, state) => state.Invoke());
                 }
 
-                private void Invoke()
+                private IDisposable Invoke()
                 {
                     ForwardOnNext(0);
                     ForwardOnCompleted();
+                    return Disposable.Empty;
                 }
             }
         }
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/QueryLanguage.Async.cs b/Rx.NET/Source/src/System.Reactive/Linq/QueryLanguage.Async.cs
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/QueryLanguageEx.cs b/Rx.NET/Source/src/System.Reactive/Linq/QueryLanguageEx.cs

Original file line number	Diff line number	Diff line change
`@@ -434,7 +434,7 @@ private void DrainQueue(ICancelable cancel)`
`434`	`434`	`if (shouldWait)`
`435`	`435`	`{`
`436`	`436`	`var timer = new ManualResetEventSlim();`
`437`		`- _scheduler.Schedule(waitTime, () => { timer.Set(); });`
	`437`	`+ _scheduler.Schedule(timer, waitTime, (_, slimTimer) => { slimTimer.Set(); return Disposable.Empty; });`
`438`	`438`
`439`	`439`	`try`
`440`	`440`	`{`
Original file line number	Diff line number	Diff line change
`@@ -284,12 +284,13 @@ public IDisposable Connect(IObserver<TArgs> observer)`
`284`	`284`	`{`
`285`	`285`	`if (--_count == 0)`
`286`	`286`	`{`
`287`		`- _parent._scheduler.Schedule(_removeHandler.Dispose);`
	`287`	`+ _parent._scheduler.ScheduleAction(_removeHandler, handler => handler.Dispose());`
`288`	`288`	`_parent._session = null;`
`289`	`289`	`}`
`290`	`290`	`}`
`291`	`291`	`});`
`292`	`292`	`}`
	`293`	`+`
`293`	`294`	`}`
`294`	`295`
`295`	`296`	`private void Initialize()`
Original file line number	Diff line number	Diff line change
`@@ -3,6 +3,7 @@`
`3`	`3`	`// See the LICENSE file in the project root for more information.`
`4`	`4`
`5`	`5`	`using System.Reactive.Concurrency;`
	`6`	`+using System.Reactive.Disposables;`
`6`	`7`
`7`	`8`	`namespace System.Reactive.Linq.ObservableImpl`
`8`	`9`	`{`
`@@ -33,7 +34,7 @@ public _(TResult value, IObserver<TResult> observer, IDisposable cancel)`
`33`	`34`
`34`	`35`	`public IDisposable Run(IScheduler scheduler)`
`35`	`36`	`{`
`36`		`- return scheduler.Schedule(Invoke);`
	`37`	`+ return scheduler.ScheduleAction(this, @this => @this.Invoke());`
`37`	`38`	`}`
`38`	`39`
`39`	`40`	`private void Invoke()`
Original file line number	Diff line number	Diff line change
`@@ -101,14 +101,15 @@ public _(IObserver<TSource> observer, IDisposable cancel)`
`101`	`101`
`102`	`102`	`public IDisposable Run(Time parent)`
`103`	`103`	`{`
`104`		`- var t = parent._scheduler.Schedule(parent._duration, Tick);`
	`104`	`+ var t = parent._scheduler.Schedule(this, parent._duration, (_, state) => state.Tick());`
`105`	`105`	`var d = parent._source.SubscribeSafe(this);`
`106`	`106`	`return StableCompositeDisposable.Create(t, d);`
`107`	`107`	`}`
`108`	`108`
`109`		`- private void Tick()`
	`109`	`+ private IDisposable Tick()`
`110`	`110`	`{`
`111`	`111`	`_open = true;`
	`112`	`+ return Disposable.Empty;`
`112`	`113`	`}`
`113`	`114`
`114`	`115`	`public override void OnNext(TSource value)`
Original file line number	Diff line number	Diff line change
`@@ -205,14 +205,15 @@ public _(IObserver<TSource> observer, IDisposable cancel)`
`205`	`205`
`206`	`206`	`public IDisposable Run(SkipUntil<TSource> parent)`
`207`	`207`	`{`
`208`		`- var t = parent._scheduler.Schedule(parent._startTime, Tick);`
	`208`	`+ var t = parent._scheduler.Schedule(this, parent._startTime, (_, state) => state.Tick());`
`209`	`209`	`var d = parent._source.SubscribeSafe(this);`
`210`	`210`	`return StableCompositeDisposable.Create(t, d);`
`211`	`211`	`}`
`212`	`212`
`213`		`- private void Tick()`
	`213`	`+ private IDisposable Tick()`
`214`	`214`	`{`
`215`	`215`	`_open = true;`
	`216`	`+ return Disposable.Empty;`
`216`	`217`	`}`
`217`	`218`
`218`	`219`	`public override void OnNext(TSource value)`
Original file line number	Diff line number	Diff line change
`@@ -112,17 +112,18 @@ public IDisposable Run(Time parent)`
`112`	`112`	`{`
`113`	`113`	`_gate = new object();`
`114`	`114`
`115`		`- var t = parent._scheduler.Schedule(parent._duration, Tick);`
	`115`	`+ var t = parent._scheduler.Schedule(this, parent._duration, (_, state) => state.Tick());`
`116`	`116`	`var d = parent._source.SubscribeSafe(this);`
`117`	`117`	`return StableCompositeDisposable.Create(t, d);`
`118`	`118`	`}`
`119`	`119`
`120`		`- private void Tick()`
	`120`	`+ private IDisposable Tick()`
`121`	`121`	`{`
`122`	`122`	`lock (_gate)`
`123`	`123`	`{`
`124`	`124`	`ForwardOnCompleted();`
`125`	`125`	`}`
	`126`	`+ return Disposable.Empty;`
`126`	`127`	`}`
`127`	`128`
`128`	`129`	`public override void OnNext(TSource value)`
Original file line number	Diff line number	Diff line change
`@@ -180,17 +180,18 @@ public _(IObserver<TSource> observer, IDisposable cancel)`
`180`	`180`
`181`	`181`	`public IDisposable Run(TakeUntil<TSource> parent)`
`182`	`182`	`{`
`183`		`- var t = parent._scheduler.Schedule(parent._endTime, Tick);`
	`183`	`+ var t = parent._scheduler.Schedule(this, parent._endTime, (_, state) => state.Tick());`
`184`	`184`	`var d = parent._source.SubscribeSafe(this);`
`185`	`185`	`return StableCompositeDisposable.Create(t, d);`
`186`	`186`	`}`
`187`	`187`
`188`		`- private void Tick()`
	`188`	`+ private IDisposable Tick()`
`189`	`189`	`{`
`190`	`190`	`lock (_gate)`
`191`	`191`	`{`
`192`	`192`	`ForwardOnCompleted();`
`193`	`193`	`}`
	`194`	`+ return Disposable.Empty;`
`194`	`195`	`}`
`195`	`196`
`196`	`197`	`public override void OnNext(TSource value)`
Original file line number	Diff line number	Diff line change
`@@ -173,14 +173,14 @@ public IDisposable Run(Absolute parent)`
`173`	`173`
`174`	`174`	`_switched = false;`
`175`	`175`
`176`		`- var timer = parent._scheduler.Schedule(parent._dueTime, Timeout);`
	`176`	`+ var timer = parent._scheduler.Schedule(this, parent._dueTime, (_, state) => state.Timeout());`
`177`	`177`
`178`	`178`	`original.Disposable = parent._source.SubscribeSafe(this);`
`179`	`179`
`180`	`180`	`return StableCompositeDisposable.Create(_subscription, timer);`
`181`	`181`	`}`
`182`	`182`
`183`		`- private void Timeout()`
	`183`	`+ private IDisposable Timeout()`
`184`	`184`	`{`
`185`	`185`	`var timerWins = false;`
`186`	`186`
`@@ -192,6 +192,8 @@ private void Timeout()`
`192`	`192`
`193`	`193`	`if (timerWins)`
`194`	`194`	`_subscription.Disposable = _other.SubscribeSafe(GetForwarder());`
	`195`	`+`
	`196`	`+ return Disposable.Empty;`
`195`	`197`	`}`
`196`	`198`
`197`	`199`	`public override void OnNext(TSource value)`