Inline CancellationDisposable in SelectMany.

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

@@ -534,7 +534,7 @@ namespace System.Reactive.Linq.ObservableImpl
             internal sealed class _ : Sink<TSource, TResult>
             {
                 private readonly object _gate = new object();
-                private readonly CancellationDisposable _cancel = new CancellationDisposable();
+                private readonly CancellationTokenSource _cancel = new CancellationTokenSource();
 
                 private readonly Func<TSource, CancellationToken, Task<TCollection>> _collectionSelector;
                 private readonly Func<TSource, TCollection, TResult> _resultSelector;
@@ -559,7 +559,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     if (disposing)
                     {
-                        _cancel.Dispose();
+                        _cancel.Cancel();
                     }
                     base.Dispose(disposing);
                 }
@@ -639,7 +639,7 @@ namespace System.Reactive.Linq.ObservableImpl
                         break;
                         case TaskStatus.Canceled:
                         {
-                            if (!_cancel.IsDisposed)
+                            if (!_cancel.IsCancellationRequested)
                             {
                                 lock (_gate)
                                 {
@@ -692,7 +692,7 @@ namespace System.Reactive.Linq.ObservableImpl
             internal sealed class _ : Sink<TSource, TResult>
             {
                 private readonly object _gate = new object();
-                private readonly CancellationDisposable _cancel = new CancellationDisposable();
+                private readonly CancellationTokenSource _cancel = new CancellationTokenSource();
 
                 private readonly Func<TSource, int, CancellationToken, Task<TCollection>> _collectionSelector;
                 private readonly Func<TSource, int, TCollection, TResult> _resultSelector;
@@ -718,7 +718,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     if (disposing)
                     {
-                        _cancel.Dispose();
+                        _cancel.Cancel();
                     }
                     base.Dispose(disposing);
                 }
@@ -800,7 +800,7 @@ namespace System.Reactive.Linq.ObservableImpl
                         break;
                         case TaskStatus.Canceled:
                         {
-                            if (!_cancel.IsDisposed)
+                            if (!_cancel.IsCancellationRequested)
                             {
                                 lock (_gate)
                                 {