Rework SelectMany (#604)

danielcweber · web-flow · commit ad99fb5dfbb5 · 2018-06-19T13:12:03.000+02:00
diff --git a/Rx.NET/Source/src/System.Reactive/Internal/Sink.cs b/Rx.NET/Source/src/System.Reactive/Internal/Sink.cs
@@ -56,6 +56,11 @@ protected void SetUpstream(IDisposable upstream)
         {
             Disposable.SetSingle(ref _upstream, upstream);
         }
+
+        protected void DisposeUpstream()
+        {
+            Disposable.TryDispose(ref _upstream);
+        }
     }
 
     /// <summary>
diff --git a/Rx.NET/Source/src/System.Reactive/Linq/Observable/SelectMany.cs b/Rx.NET/Source/src/System.Reactive/Linq/Observable/SelectMany.cs
@@ -856,28 +856,15 @@ public ObservableSelector(IObservable<TSource> source, Func<TSource, IObservable
             internal class _ : Sink<TSource, TResult> 
             {
                 protected readonly object _gate = new object();
-                private readonly SingleAssignmentDisposable _sourceSubscription = new SingleAssignmentDisposable();
-                private readonly CompositeDisposable _group = new CompositeDisposable();
-
                 private readonly Func<TSource, IObservable<TResult>> _selector;
+                private readonly CompositeDisposable _group = new CompositeDisposable();
+                
+                private bool _isStopped;
 
                 public _(ObservableSelector parent, IObserver<TResult> observer)
                     : base(observer)
                 {
                     _selector = parent._selector;
-
-                    _group.Add(_sourceSubscription);
-                }
-
-                private bool _isStopped;
-
-                public override void Run(IObservable<TSource> source)
-                {
-                    _isStopped = false;
-
-                    _sourceSubscription.Disposable = source.SubscribeSafe(this);
-
-                    SetUpstream(_group);
                 }
 
                 public override void OnNext(TSource value)
@@ -913,14 +900,22 @@ public override void OnCompleted()
                     Final();
                 }
 
+                protected override void Dispose(bool disposing)
+                {
+                    base.Dispose(disposing);
+
+                    if (disposing)
+                        _group.Dispose();
+                }
+
                 protected void Final()
                 {
                     _isStopped = true;
-                    if (_group.Count == 1)
+                    if (_group.Count == 0)
                     {
                         //
                         // Notice there can be a race between OnCompleted of the source and any
-                        // of the inner sequences, where both see _group.Count == 1, and one is
+                        // of the inner sequences, where both see _group.Count == 0, and one is
                         // waiting for the lock. There won't be a double OnCompleted observation
                         // though, because the call to Dispose silences the observer by swapping
                         // in a NopObserver<T>.
@@ -932,46 +927,45 @@ protected void Final()
                     }
                     else
                     {
-                        _sourceSubscription.Dispose();
+                        DisposeUpstream();
                     }
                 }
 
                 protected void SubscribeInner(IObservable<TResult> inner)
                 {
-                    var innerSubscription = new SingleAssignmentDisposable();
-                    _group.Add(innerSubscription);
-                    innerSubscription.Disposable = inner.SubscribeSafe(new InnerObserver(this, innerSubscription));
+                    var innerObserver = new InnerObserver(this);
+
+                    _group.Add(innerObserver);
+                    innerObserver.SetResource(inner.SubscribeSafe(innerObserver));
                 }
 
-                private sealed class InnerObserver : IObserver<TResult>
+                private sealed class InnerObserver : SafeObserver<TResult>
                 {
                     private readonly _ _parent;
-                    private readonly IDisposable _self;
 
-                    public InnerObserver(_ parent, IDisposable self)
+                    public InnerObserver(_ parent)
                     {
                         _parent = parent;
-                        _self = self;
                     }
 
-                    public void OnNext(TResult value)
+                    public override void OnNext(TResult value)
                     {
                         lock (_parent._gate)
                             _parent.ForwardOnNext(value);
                     }
 
-                    public void OnError(Exception error)
+                    public override void OnError(Exception error)
                     {
                         lock (_parent._gate)
                         {
                             _parent.ForwardOnError(error);
                         }
                     }
 
-                    public void OnCompleted()
+                    public override void OnCompleted()
                     {
-                        _parent._group.Remove(_self);
-                        if (_parent._isStopped && _parent._group.Count == 1)
+                        _parent._group.Remove(this);
+                        if (_parent._isStopped && _parent._group.Count == 0)
                         {
                             //
                             // Notice there can be a race between OnCompleted of the source and any
@@ -1090,29 +1084,17 @@ public ObservableSelectorIndexed(IObservable<TSource> source, Func<TSource, int,
             internal class _ : Sink<TSource, TResult> 
             {
                 private readonly object _gate = new object();
-                private readonly SingleAssignmentDisposable _sourceSubscription = new SingleAssignmentDisposable();
                 private readonly CompositeDisposable _group = new CompositeDisposable();
 
                 protected readonly Func<TSource, int, IObservable<TResult>> _selector;
 
+                private int _index;
+                private bool _isStopped;
+
                 public _(ObservableSelectorIndexed parent, IObserver<TResult> observer)
                     : base(observer)
                 {
                     _selector = parent._selector;
-
-                    _group.Add(_sourceSubscription);
-                }
-
-                private bool _isStopped;
-                private int _index;
-
-                public override void Run(IObservable<TSource> source)
-                {
-                    _isStopped = false;
-
-                    _sourceSubscription.Disposable = source.SubscribeSafe(this);
-
-                    SetUpstream(_group);
                 }
 
                 public override void OnNext(TSource value)
@@ -1148,10 +1130,18 @@ public override void OnCompleted()
                     Final();
                 }
 
+                protected override void Dispose(bool disposing)
+                {
+                    base.Dispose(disposing);
+
+                    if (disposing)
+                        _group.Dispose();
+                }
+
                 protected void Final()
                 {
                     _isStopped = true;
-                    if (_group.Count == 1)
+                    if (_group.Count == 0)
                     {
                         //
                         // Notice there can be a race between OnCompleted of the source and any
@@ -1167,46 +1157,45 @@ protected void Final()
                     }
                     else
                     {
-                        _sourceSubscription.Dispose();
+                        DisposeUpstream();
                     }
                 }
 
                 protected void SubscribeInner(IObservable<TResult> inner)
                 {
-                    var innerSubscription = new SingleAssignmentDisposable();
-                    _group.Add(innerSubscription);
-                    innerSubscription.Disposable = inner.SubscribeSafe(new InnerObserver(this, innerSubscription));
+                    var innerObserver = new InnerObserver(this);
+
+                    _group.Add(innerObserver);
+                    innerObserver.SetResource(inner.SubscribeSafe(innerObserver));
                 }
 
-                private sealed class InnerObserver : IObserver<TResult>
+                private sealed class InnerObserver : SafeObserver<TResult>
                 {
                     private readonly _ _parent;
-                    private readonly IDisposable _self;
 
-                    public InnerObserver(_ parent, IDisposable self)
+                    public InnerObserver(_ parent)
                     {
                         _parent = parent;
-                        _self = self;
                     }
 
-                    public void OnNext(TResult value)
+                    public override void OnNext(TResult value)
                     {
                         lock (_parent._gate)
                             _parent.ForwardOnNext(value);
                     }
 
-                    public void OnError(Exception error)
+                    public override void OnError(Exception error)
                     {
                         lock (_parent._gate)
                         {
                             _parent.ForwardOnError(error);
                         }
                     }
 
-                    public void OnCompleted()
+                    public override void OnCompleted()
                     {
-                        _parent._group.Remove(_self);
-                        if (_parent._isStopped && _parent._group.Count == 1)
+                        _parent._group.Remove(this);
+                        if (_parent._isStopped && _parent._group.Count == 0)
                         {
                             //
                             // Notice there can be a race between OnCompleted of the source and any
@@ -1242,8 +1231,6 @@ public ObservableSelectorsIndexed(IObservable<TSource> source, Func<TSource, int
             new internal sealed class _ : ObservableSelectorIndexed._
             {
                 private readonly object _gate = new object();
-                private readonly SingleAssignmentDisposable _sourceSubscription = new SingleAssignmentDisposable();
-                private readonly CompositeDisposable _group = new CompositeDisposable();
 
                 private readonly Func<Exception, IObservable<TResult>> _selectorOnError;
                 private readonly Func<IObservable<TResult>> _selectorOnCompleted;
@@ -1253,8 +1240,6 @@ public _(ObservableSelectorsIndexed parent, IObserver<TResult> observer)
                 {
                     _selectorOnError = parent._selectorOnError;
                     _selectorOnCompleted = parent._selectorOnCompleted;
-
-                    _group.Add(_sourceSubscription);
                 }
 
                 public override void OnError(Exception error)
@@ -1497,7 +1482,7 @@ public TaskSelector(IObservable<TSource> source, Func<TSource, CancellationToken
             internal sealed class _ : Sink<TSource, TResult> 
             {
                 private readonly object _gate = new object();
-                private readonly CancellationDisposable _cancel = new CancellationDisposable();
+                private readonly CancellationTokenSource _cts = new CancellationTokenSource();
 
                 private readonly Func<TSource, CancellationToken, Task<TResult>> _selector;
 
@@ -1520,7 +1505,7 @@ protected override void Dispose(bool disposing)
                 {
                     if (disposing)
                     {
-                        _cancel.Dispose();
+                        _cts.Cancel();
                     }
                     base.Dispose(disposing);
                 }
@@ -1531,7 +1516,7 @@ public override void OnNext(TSource value)
                     try
                     {
                         Interlocked.Increment(ref _count);
-                        task = _selector(value, _cancel.Token);
+                        task = _selector(value, _cts.Token);
                     }
                     catch (Exception ex)
                     {
@@ -1549,7 +1534,7 @@ public override void OnNext(TSource value)
                     }
                     else
                     {
-                        task.ContinueWith(OnCompletedTask);
+                        task.ContinueWith((closureTask, @thisObject) => ((_)@thisObject).OnCompletedTask(closureTask), this);
                     }
                 }
 
@@ -1575,7 +1560,7 @@ private void OnCompletedTask(Task<TResult> task)
                             break;
                         case TaskStatus.Canceled:
                             {
-                                if (!_cancel.IsDisposed)
+                                if (!_cts.IsCancellationRequested)
                                 {
                                     lock (_gate)
                                     {
@@ -1626,7 +1611,7 @@ public TaskSelectorIndexed(IObservable<TSource> source, Func<TSource, int, Cance
             internal sealed class _ : Sink<TSource, TResult> 
             {
                 private readonly object _gate = new object();
-                private readonly CancellationDisposable _cancel = new CancellationDisposable();
+                private readonly CancellationTokenSource _cts = new CancellationTokenSource();
 
                 private readonly Func<TSource, int, CancellationToken, Task<TResult>> _selector;
 
@@ -1650,7 +1635,7 @@ protected override void Dispose(bool disposing)
                 {
                     if (disposing)
                     {
-                        _cancel.Dispose();
+                        _cts.Cancel();
                     }
                     base.Dispose(disposing);
                 }
@@ -1661,7 +1646,7 @@ public override void OnNext(TSource value)
                     try
                     {
                         Interlocked.Increment(ref _count);
-                        task = _selector(value, checked(_index++), _cancel.Token);
+                        task = _selector(value, checked(_index++), _cts.Token);
                     }
                     catch (Exception ex)
                     {
@@ -1679,7 +1664,7 @@ public override void OnNext(TSource value)
                     }
                     else
                     {
-                        task.ContinueWith(OnCompletedTask);
+                        task.ContinueWith((closureTask, @thisObject) => ((_)@thisObject).OnCompletedTask(closureTask), this);
                     }
                 }
 
@@ -1705,7 +1690,7 @@ private void OnCompletedTask(Task<TResult> task)
                             break;
                         case TaskStatus.Canceled:
                             {
-                                if (!_cancel.IsDisposed)
+                                if (!_cts.IsCancellationRequested)
                                 {
                                     lock (_gate)
                                     {

Original file line number	Diff line number	Diff line change
`@@ -56,6 +56,11 @@ protected void SetUpstream(IDisposable upstream)`
`56`	`56`	`{`
`57`	`57`	`Disposable.SetSingle(ref _upstream, upstream);`
`58`	`58`	`}`
	`59`	`+`
	`60`	`+ protected void DisposeUpstream()`
	`61`	`+ {`
	`62`	`+ Disposable.TryDispose(ref _upstream);`
	`63`	`+ }`
`59`	`64`	`}`
`60`	`65`
`61`	`66`	`/// <summary>`
Original file line number	Diff line number	Diff line change
`@@ -856,28 +856,15 @@ public ObservableSelector(IObservable<TSource> source, Func<TSource, IObservable`
`856`	`856`	`internal class _ : Sink<TSource, TResult>`
`857`	`857`	`{`
`858`	`858`	`protected readonly object _gate = new object();`
`859`		`- private readonly SingleAssignmentDisposable _sourceSubscription = new SingleAssignmentDisposable();`
`860`		`- private readonly CompositeDisposable _group = new CompositeDisposable();`
`861`		`-`
`862`	`859`	`private readonly Func<TSource, IObservable<TResult>> _selector;`
	`860`	`+ private readonly CompositeDisposable _group = new CompositeDisposable();`
	`861`	`+`
	`862`	`+ private bool _isStopped;`
`863`	`863`
`864`	`864`	`public _(ObservableSelector parent, IObserver<TResult> observer)`
`865`	`865`	`: base(observer)`
`866`	`866`	`{`
`867`	`867`	`_selector = parent._selector;`
`868`		`-`
`869`		`- _group.Add(_sourceSubscription);`
`870`		`- }`
`871`		`-`
`872`		`- private bool _isStopped;`
`873`		`-`
`874`		`- public override void Run(IObservable<TSource> source)`
`875`		`- {`
`876`		`- _isStopped = false;`
`877`		`-`
`878`		`- _sourceSubscription.Disposable = source.SubscribeSafe(this);`
`879`		`-`
`880`		`- SetUpstream(_group);`
`881`	`868`	`}`
`882`	`869`
`883`	`870`	`public override void OnNext(TSource value)`
`@@ -913,14 +900,22 @@ public override void OnCompleted()`
`913`	`900`	`Final();`
`914`	`901`	`}`
`915`	`902`
	`903`	`+ protected override void Dispose(bool disposing)`
	`904`	`+ {`
	`905`	`+ base.Dispose(disposing);`
	`906`	`+`
	`907`	`+ if (disposing)`
	`908`	`+ _group.Dispose();`
	`909`	`+ }`
	`910`	`+`
`916`	`911`	`protected void Final()`
`917`	`912`	`{`
`918`	`913`	`_isStopped = true;`
`919`		`- if (_group.Count == 1)`
	`914`	`+ if (_group.Count == 0)`
`920`	`915`	`{`
`921`	`916`	`//`
`922`	`917`	`// Notice there can be a race between OnCompleted of the source and any`
`923`		`- // of the inner sequences, where both see _group.Count == 1, and one is`
	`918`	`+ // of the inner sequences, where both see _group.Count == 0, and one is`
`924`	`919`	`// waiting for the lock. There won't be a double OnCompleted observation`
`925`	`920`	`// though, because the call to Dispose silences the observer by swapping`
`926`	`921`	`// in a NopObserver<T>.`
`@@ -932,46 +927,45 @@ protected void Final()`
`932`	`927`	`}`
`933`	`928`	`else`
`934`	`929`	`{`
`935`		`- _sourceSubscription.Dispose();`
	`930`	`+ DisposeUpstream();`
`936`	`931`	`}`
`937`	`932`	`}`
`938`	`933`
`939`	`934`	`protected void SubscribeInner(IObservable<TResult> inner)`
`940`	`935`	`{`
`941`		`- var innerSubscription = new SingleAssignmentDisposable();`
`942`		`- _group.Add(innerSubscription);`
`943`		`- innerSubscription.Disposable = inner.SubscribeSafe(new InnerObserver(this, innerSubscription));`
	`936`	`+ var innerObserver = new InnerObserver(this);`
	`937`	`+`
	`938`	`+ _group.Add(innerObserver);`
	`939`	`+ innerObserver.SetResource(inner.SubscribeSafe(innerObserver));`
`944`	`940`	`}`
`945`	`941`
`946`		`- private sealed class InnerObserver : IObserver<TResult>`
	`942`	`+ private sealed class InnerObserver : SafeObserver<TResult>`
`947`	`943`	`{`
`948`	`944`	`private readonly _ _parent;`
`949`		`- private readonly IDisposable _self;`
`950`	`945`
`951`		`- public InnerObserver(_ parent, IDisposable self)`
	`946`	`+ public InnerObserver(_ parent)`
`952`	`947`	`{`
`953`	`948`	`_parent = parent;`
`954`		`- _self = self;`
`955`	`949`	`}`
`956`	`950`
`957`		`- public void OnNext(TResult value)`
	`951`	`+ public override void OnNext(TResult value)`
`958`	`952`	`{`
`959`	`953`	`lock (_parent._gate)`
`960`	`954`	`_parent.ForwardOnNext(value);`
`961`	`955`	`}`
`962`	`956`
`963`		`- public void OnError(Exception error)`
	`957`	`+ public override void OnError(Exception error)`
`964`	`958`	`{`
`965`	`959`	`lock (_parent._gate)`
`966`	`960`	`{`
`967`	`961`	`_parent.ForwardOnError(error);`
`968`	`962`	`}`
`969`	`963`	`}`
`970`	`964`
`971`		`- public void OnCompleted()`
	`965`	`+ public override void OnCompleted()`
`972`	`966`	`{`
`973`		`- _parent._group.Remove(_self);`
`974`		`- if (_parent._isStopped && _parent._group.Count == 1)`
	`967`	`+ _parent._group.Remove(this);`
	`968`	`+ if (_parent._isStopped && _parent._group.Count == 0)`
`975`	`969`	`{`
`976`	`970`	`//`
`977`	`971`	`// Notice there can be a race between OnCompleted of the source and any`
`@@ -1090,29 +1084,17 @@ public ObservableSelectorIndexed(IObservable<TSource> source, Func<TSource, int,`
`1090`	`1084`	`internal class _ : Sink<TSource, TResult>`
`1091`	`1085`	`{`
`1092`	`1086`	`private readonly object _gate = new object();`
`1093`		`- private readonly SingleAssignmentDisposable _sourceSubscription = new SingleAssignmentDisposable();`
`1094`	`1087`	`private readonly CompositeDisposable _group = new CompositeDisposable();`
`1095`	`1088`
`1096`	`1089`	`protected readonly Func<TSource, int, IObservable<TResult>> _selector;`
`1097`	`1090`
	`1091`	`+ private int _index;`
	`1092`	`+ private bool _isStopped;`
	`1093`	`+`
`1098`	`1094`	`public _(ObservableSelectorIndexed parent, IObserver<TResult> observer)`
`1099`	`1095`	`: base(observer)`
`1100`	`1096`	`{`
`1101`	`1097`	`_selector = parent._selector;`
`1102`		`-`
`1103`		`- _group.Add(_sourceSubscription);`
`1104`		`- }`
`1105`		`-`
`1106`		`- private bool _isStopped;`
`1107`		`- private int _index;`
`1108`		`-`
`1109`		`- public override void Run(IObservable<TSource> source)`
`1110`		`- {`
`1111`		`- _isStopped = false;`
`1112`		`-`
`1113`		`- _sourceSubscription.Disposable = source.SubscribeSafe(this);`
`1114`		`-`
`1115`		`- SetUpstream(_group);`
`1116`	`1098`	`}`
`1117`	`1099`
`1118`	`1100`	`public override void OnNext(TSource value)`
`@@ -1148,10 +1130,18 @@ public override void OnCompleted()`
`1148`	`1130`	`Final();`
`1149`	`1131`	`}`
`1150`	`1132`
	`1133`	`+ protected override void Dispose(bool disposing)`
	`1134`	`+ {`
	`1135`	`+ base.Dispose(disposing);`
	`1136`	`+`
	`1137`	`+ if (disposing)`
	`1138`	`+ _group.Dispose();`
	`1139`	`+ }`
	`1140`	`+`
`1151`	`1141`	`protected void Final()`
`1152`	`1142`	`{`
`1153`	`1143`	`_isStopped = true;`
`1154`		`- if (_group.Count == 1)`
	`1144`	`+ if (_group.Count == 0)`
`1155`	`1145`	`{`
`1156`	`1146`	`//`
`1157`	`1147`	`// Notice there can be a race between OnCompleted of the source and any`
`@@ -1167,46 +1157,45 @@ protected void Final()`
`1167`	`1157`	`}`
`1168`	`1158`	`else`
`1169`	`1159`	`{`
`1170`		`- _sourceSubscription.Dispose();`
	`1160`	`+ DisposeUpstream();`
`1171`	`1161`	`}`
`1172`	`1162`	`}`
`1173`	`1163`
`1174`	`1164`	`protected void SubscribeInner(IObservable<TResult> inner)`
`1175`	`1165`	`{`
`1176`		`- var innerSubscription = new SingleAssignmentDisposable();`
`1177`		`- _group.Add(innerSubscription);`
`1178`		`- innerSubscription.Disposable = inner.SubscribeSafe(new InnerObserver(this, innerSubscription));`
	`1166`	`+ var innerObserver = new InnerObserver(this);`
	`1167`	`+`
	`1168`	`+ _group.Add(innerObserver);`
	`1169`	`+ innerObserver.SetResource(inner.SubscribeSafe(innerObserver));`
`1179`	`1170`	`}`
`1180`	`1171`
`1181`		`- private sealed class InnerObserver : IObserver<TResult>`
	`1172`	`+ private sealed class InnerObserver : SafeObserver<TResult>`
`1182`	`1173`	`{`
`1183`	`1174`	`private readonly _ _parent;`
`1184`		`- private readonly IDisposable _self;`
`1185`	`1175`
`1186`		`- public InnerObserver(_ parent, IDisposable self)`
	`1176`	`+ public InnerObserver(_ parent)`
`1187`	`1177`	`{`
`1188`	`1178`	`_parent = parent;`
`1189`		`- _self = self;`
`1190`	`1179`	`}`
`1191`	`1180`
`1192`		`- public void OnNext(TResult value)`
	`1181`	`+ public override void OnNext(TResult value)`
`1193`	`1182`	`{`
`1194`	`1183`	`lock (_parent._gate)`
`1195`	`1184`	`_parent.ForwardOnNext(value);`
`1196`	`1185`	`}`
`1197`	`1186`
`1198`		`- public void OnError(Exception error)`
	`1187`	`+ public override void OnError(Exception error)`
`1199`	`1188`	`{`
`1200`	`1189`	`lock (_parent._gate)`
`1201`	`1190`	`{`
`1202`	`1191`	`_parent.ForwardOnError(error);`
`1203`	`1192`	`}`
`1204`	`1193`	`}`
`1205`	`1194`
`1206`		`- public void OnCompleted()`
	`1195`	`+ public override void OnCompleted()`
`1207`	`1196`	`{`
`1208`		`- _parent._group.Remove(_self);`
`1209`		`- if (_parent._isStopped && _parent._group.Count == 1)`
	`1197`	`+ _parent._group.Remove(this);`
	`1198`	`+ if (_parent._isStopped && _parent._group.Count == 0)`
`1210`	`1199`	`{`
`1211`	`1200`	`//`
`1212`	`1201`	`// Notice there can be a race between OnCompleted of the source and any`
`@@ -1242,8 +1231,6 @@ public ObservableSelectorsIndexed(IObservable<TSource> source, Func<TSource, int`
`1242`	`1231`	`new internal sealed class _ : ObservableSelectorIndexed._`
`1243`	`1232`	`{`
`1244`	`1233`	`private readonly object _gate = new object();`
`1245`		`- private readonly SingleAssignmentDisposable _sourceSubscription = new SingleAssignmentDisposable();`
`1246`		`- private readonly CompositeDisposable _group = new CompositeDisposable();`
`1247`	`1234`
`1248`	`1235`	`private readonly Func<Exception, IObservable<TResult>> _selectorOnError;`
`1249`	`1236`	`private readonly Func<IObservable<TResult>> _selectorOnCompleted;`
`@@ -1253,8 +1240,6 @@ public _(ObservableSelectorsIndexed parent, IObserver<TResult> observer)`
`1253`	`1240`	`{`
`1254`	`1241`	`_selectorOnError = parent._selectorOnError;`
`1255`	`1242`	`_selectorOnCompleted = parent._selectorOnCompleted;`
`1256`		`-`
`1257`		`- _group.Add(_sourceSubscription);`
`1258`	`1243`	`}`
`1259`	`1244`
`1260`	`1245`	`public override void OnError(Exception error)`
`@@ -1497,7 +1482,7 @@ public TaskSelector(IObservable<TSource> source, Func<TSource, CancellationToken`
`1497`	`1482`	`internal sealed class _ : Sink<TSource, TResult>`
`1498`	`1483`	`{`
`1499`	`1484`	`private readonly object _gate = new object();`
`1500`		`- private readonly CancellationDisposable _cancel = new CancellationDisposable();`
	`1485`	`+ private readonly CancellationTokenSource _cts = new CancellationTokenSource();`
`1501`	`1486`
`1502`	`1487`	`private readonly Func<TSource, CancellationToken, Task<TResult>> _selector;`
`1503`	`1488`
`@@ -1520,7 +1505,7 @@ protected override void Dispose(bool disposing)`
`1520`	`1505`	`{`
`1521`	`1506`	`if (disposing)`
`1522`	`1507`	`{`
`1523`		`- _cancel.Dispose();`
	`1508`	`+ _cts.Cancel();`
`1524`	`1509`	`}`
`1525`	`1510`	`base.Dispose(disposing);`
`1526`	`1511`	`}`
`@@ -1531,7 +1516,7 @@ public override void OnNext(TSource value)`
`1531`	`1516`	`try`
`1532`	`1517`	`{`
`1533`	`1518`	`Interlocked.Increment(ref _count);`
`1534`		`- task = _selector(value, _cancel.Token);`
	`1519`	`+ task = _selector(value, _cts.Token);`
`1535`	`1520`	`}`
`1536`	`1521`	`catch (Exception ex)`
`1537`	`1522`	`{`
`@@ -1549,7 +1534,7 @@ public override void OnNext(TSource value)`
`1549`	`1534`	`}`
`1550`	`1535`	`else`
`1551`	`1536`	`{`
`1552`		`- task.ContinueWith(OnCompletedTask);`
	`1537`	`+ task.ContinueWith((closureTask, @thisObject) => ((_)@thisObject).OnCompletedTask(closureTask), this);`
`1553`	`1538`	`}`
`1554`	`1539`	`}`
`1555`	`1540`
`@@ -1575,7 +1560,7 @@ private void OnCompletedTask(Task<TResult> task)`
`1575`	`1560`	`break;`
`1576`	`1561`	`case TaskStatus.Canceled:`
`1577`	`1562`	`{`
`1578`		`- if (!_cancel.IsDisposed)`
	`1563`	`+ if (!_cts.IsCancellationRequested)`
`1579`	`1564`	`{`
`1580`	`1565`	`lock (_gate)`
`1581`	`1566`	`{`
`@@ -1626,7 +1611,7 @@ public TaskSelectorIndexed(IObservable<TSource> source, Func<TSource, int, Cance`
`1626`	`1611`	`internal sealed class _ : Sink<TSource, TResult>`
`1627`	`1612`	`{`
`1628`	`1613`	`private readonly object _gate = new object();`
`1629`		`- private readonly CancellationDisposable _cancel = new CancellationDisposable();`
	`1614`	`+ private readonly CancellationTokenSource _cts = new CancellationTokenSource();`
`1630`	`1615`
`1631`	`1616`	`private readonly Func<TSource, int, CancellationToken, Task<TResult>> _selector;`
`1632`	`1617`
`@@ -1650,7 +1635,7 @@ protected override void Dispose(bool disposing)`
`1650`	`1635`	`{`
`1651`	`1636`	`if (disposing)`
`1652`	`1637`	`{`
`1653`		`- _cancel.Dispose();`
	`1638`	`+ _cts.Cancel();`
`1654`	`1639`	`}`
`1655`	`1640`	`base.Dispose(disposing);`
`1656`	`1641`	`}`
`@@ -1661,7 +1646,7 @@ public override void OnNext(TSource value)`
`1661`	`1646`	`try`
`1662`	`1647`	`{`
`1663`	`1648`	`Interlocked.Increment(ref _count);`
`1664`		`- task = _selector(value, checked(_index++), _cancel.Token);`
	`1649`	`+ task = _selector(value, checked(_index++), _cts.Token);`
`1665`	`1650`	`}`
`1666`	`1651`	`catch (Exception ex)`
`1667`	`1652`	`{`
`@@ -1679,7 +1664,7 @@ public override void OnNext(TSource value)`
`1679`	`1664`	`}`
`1680`	`1665`	`else`
`1681`	`1666`	`{`
`1682`		`- task.ContinueWith(OnCompletedTask);`
	`1667`	`+ task.ContinueWith((closureTask, @thisObject) => ((_)@thisObject).OnCompletedTask(closureTask), this);`
`1683`	`1668`	`}`
`1684`	`1669`	`}`
`1685`	`1670`
`@@ -1705,7 +1690,7 @@ private void OnCompletedTask(Task<TResult> task)`
`1705`	`1690`	`break;`
`1706`	`1691`	`case TaskStatus.Canceled:`
`1707`	`1692`	`{`
`1708`		`- if (!_cancel.IsDisposed)`
	`1693`	`+ if (!_cts.IsCancellationRequested)`
`1709`	`1694`	`{`
`1710`	`1695`	`lock (_gate)`
`1711`	`1696`	`{`