4.x: Use Schedule calls with state

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;

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
                             {

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()

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()

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)

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)

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)

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)

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);
-            }
         }
     }
 }

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)

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;
                 }
             }
         }