More #nullable for operator implementations.

Rx.NET/Source/src/System.Reactive/Internal/TailRecursiveSink.cs

@@ -40,6 +40,7 @@ namespace System.Reactive
             {
                 DisposeAll();
             }
+
             base.Dispose(disposing);
         }
 

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 
@@ -107,7 +105,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                 }
 
-                private IDisposable _sourceDisposable;
+                private IDisposable? _sourceDisposable;
 
                 public void Run(Time parent)
                 {
@@ -121,6 +119,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     {
                         Disposable.Dispose(ref _sourceDisposable);
                     }
+
                     base.Dispose(disposing);
                 }
 

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 using System.Collections.Generic;
 using System.Reactive.Concurrency;
 
@@ -41,6 +39,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 public override void OnNext(TSource value)
                 {
                     _queue.Enqueue(value);
+
                     if (_queue.Count > _count)
                     {
                         ForwardOnNext(_queue.Dequeue());
@@ -78,7 +77,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     _queue = new Queue<Reactive.TimeInterval<TSource>>();
                 }
 
-                private IStopwatch _watch;
+                private IStopwatch? _watch;
 
                 public void Run(Time parent)
                 {
@@ -89,8 +88,10 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnNext(TSource value)
                 {
-                    var now = _watch.Elapsed;
+                    var now = _watch!.Elapsed;
+
                     _queue.Enqueue(new Reactive.TimeInterval<TSource>(value, now));
+
                     while (_queue.Count > 0 && now - _queue.Peek().Interval >= _duration)
                     {
                         ForwardOnNext(_queue.Dequeue().Value);
@@ -99,7 +100,8 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    var now = _watch.Elapsed;
+                    var now = _watch!.Elapsed;
+
                     while (_queue.Count > 0 && now - _queue.Peek().Interval >= _duration)
                     {
                         ForwardOnNext(_queue.Dequeue().Value);

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 namespace System.Reactive.Linq.ObservableImpl
 {
     internal static class SkipWhile<TSource>
@@ -25,7 +23,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal sealed class _ : IdentitySink<TSource>
             {
-                private Func<TSource, bool> _predicate;
+                private Func<TSource, bool>? _predicate;
 
                 public _(Func<TSource, bool> predicate, IObserver<TSource> observer)
                     : base(observer)
@@ -80,7 +78,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal sealed class _ : IdentitySink<TSource>
             {
-                private Func<TSource, int, bool> _predicate;
+                private Func<TSource, int, bool>? _predicate;
                 private int _index;
 
                 public _(Func<TSource, int, bool> predicate, IObserver<TSource> observer)

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 
@@ -107,19 +105,17 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal sealed class _ : IdentitySink<TSource>
             {
+                private readonly object _gate = new object();
+
                 public _(IObserver<TSource> observer)
                     : base(observer)
                 {
                 }
 
-                private object _gate;
-
-                private IDisposable _task;
+                private IDisposable? _task;
 
                 public void Run(Time parent)
                 {
-                    _gate = new object();
-
                     Disposable.SetSingle(ref _task, parent._scheduler.ScheduleAction(this, parent._duration, state => state.Tick()));
                     Run(parent._source);
                 }
@@ -130,6 +126,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     {
                         Disposable.Dispose(ref _task);
                     }
+
                     base.Dispose(disposing);
                 }
 

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 using System.Collections.Generic;
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
@@ -43,7 +41,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     _queue = new Queue<TSource>();
                 }
 
-                private IDisposable _loopDisposable;
+                private IDisposable? _loopDisposable;
 
                 protected override void Dispose(bool disposing)
                 {
@@ -152,8 +150,8 @@ namespace System.Reactive.Linq.ObservableImpl
                     _queue = new Queue<Reactive.TimeInterval<TSource>>();
                 }
 
-                private IDisposable _loopDisposable;
-                private IStopwatch _watch;
+                private IDisposable? _loopDisposable;
+                private IStopwatch? _watch;
 
                 public void Run(IObservable<TSource> source, IScheduler scheduler)
                 {
@@ -167,12 +165,13 @@ namespace System.Reactive.Linq.ObservableImpl
                     {
                         Disposable.Dispose(ref _loopDisposable);
                     }
+
                     base.Dispose(disposing);
                 }
 
                 public override void OnNext(TSource value)
                 {
-                    var now = _watch.Elapsed;
+                    var now = _watch!.Elapsed;
                     _queue.Enqueue(new Reactive.TimeInterval<TSource>(value, now));
                     Trim(now);
                 }
@@ -181,7 +180,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     DisposeUpstream();
 
-                    var now = _watch.Elapsed;
+                    var now = _watch!.Elapsed;
                     Trim(now);
 
                     var longRunning = _loopScheduler.AsLongRunning();
@@ -209,6 +208,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     {
                         ForwardOnCompleted();
                     }
+
                     return Disposable.Empty;
                 }
 

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 using System.Collections.Generic;
 using System.Reactive.Concurrency;
 
@@ -41,6 +39,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 public override void OnNext(TSource value)
                 {
                     _queue.Enqueue(value);
+
                     if (_queue.Count > _count)
                     {
                         _queue.Dequeue();
@@ -50,6 +49,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 public override void OnCompleted()
                 {
                     var res = new List<TSource>(_queue.Count);
+
                     while (_queue.Count > 0)
                     {
                         res.Add(_queue.Dequeue());
@@ -90,7 +90,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     _queue = new Queue<Reactive.TimeInterval<TSource>>();
                 }
 
-                private IStopwatch _watch;
+                private IStopwatch? _watch;
 
                 public void Run(IObservable<TSource> source, IScheduler scheduler)
                 {
@@ -101,17 +101,20 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnNext(TSource value)
                 {
-                    var now = _watch.Elapsed;
+                    var now = _watch!.Elapsed;
+
                     _queue.Enqueue(new Reactive.TimeInterval<TSource>(value, now));
+
                     Trim(now);
                 }
 
                 public override void OnCompleted()
                 {
-                    var now = _watch.Elapsed;
+                    var now = _watch!.Elapsed;
                     Trim(now);
 
                     var res = new List<TSource>(_queue.Count);
+
                     while (_queue.Count > 0)
                     {
                         res.Add(_queue.Dequeue().Value);

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 namespace System.Reactive.Linq.ObservableImpl
 {
     internal static class TakeWhile<TSource>

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

@@ -2,8 +2,6 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
-#nullable disable
-
 using System.Collections.Generic;
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
@@ -27,7 +25,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
         internal sealed class _ : IdentitySink<TSource>
         {
-            private IEnumerator<TSource> _enumerator;
+            private IEnumerator<TSource>? _enumerator;
 
             private volatile bool _disposed;
 
@@ -60,6 +58,7 @@ namespace System.Reactive.Linq.ObservableImpl
             protected override void Dispose(bool disposing)
             {
                 base.Dispose(disposing);
+
                 if (disposing)
                 {
                     _disposed = true;
@@ -72,11 +71,11 @@ namespace System.Reactive.Linq.ObservableImpl
                 var ex = default(Exception);
                 var current = default(TSource);
 
-                var enumerator = _enumerator;
+                var enumerator = _enumerator!; // NB: Loop only runs after enumerator is assigned.
 
                 if (_disposed)
                 {
-                    _enumerator.Dispose();
+                    enumerator.Dispose();
                     _enumerator = null;
 
                     return Disposable.Empty;
@@ -113,7 +112,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     return Disposable.Empty;
                 }
 
-                ForwardOnNext(current);
+                ForwardOnNext(current!); // NB: Non-null when hasNext is true.
 
                 //
                 // We never allow the scheduled work to be cancelled. Instead, the _disposed flag
@@ -199,7 +198,7 @@ namespace System.Reactive.Linq.ObservableImpl
                         break;
                     }
 
-                    ForwardOnNext(current);
+                    ForwardOnNext(current!); // NB: Non-null when hasNext is true.
                 }
 
                 enumerator.Dispose();