Enable #nullable for push-pull adapters.

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

@@ -2,7 +2,7 @@
 // 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.Diagnostics.CodeAnalysis;
 
 namespace System.Reactive.Linq.ObservableImpl
 {
@@ -37,7 +37,7 @@ namespace System.Reactive.Linq.ObservableImpl
             }
 
             private TResult _collector;
-            private Exception _error;
+            private Exception? _error;
             private bool _hasCompleted;
             private bool _done;
 
@@ -78,7 +78,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
             }
 
-            public override bool TryMoveNext(out TResult current)
+            public override bool TryMoveNext([MaybeNullWhen(false)] out TResult current)
             {
                 lock (_gate)
                 {
@@ -86,7 +86,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     if (error != null)
                     {
                         current = default;
-                        _collector = default;
+                        _collector = default!;
                         error.Throw();
                     }
                     else
@@ -96,7 +96,7 @@ namespace System.Reactive.Linq.ObservableImpl
                             if (_done)
                             {
                                 current = default;
-                                _collector = default;
+                                _collector = default!;
                                 return false;
                             }
 

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

@@ -2,8 +2,7 @@
 // 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.Diagnostics.CodeAnalysis;
 using System.Threading;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -33,8 +32,8 @@ namespace System.Reactive.Linq.ObservableImpl
 
             private bool _notificationAvailable;
             private NotificationKind _kind;
-            private TSource _value;
-            private Exception _error;
+            private TSource? _value;
+            private Exception? _error;
 
             public override void OnNext(TSource value)
             {
@@ -90,7 +89,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
             }
 
-            public override bool TryMoveNext(out TSource current)
+            public override bool TryMoveNext([MaybeNullWhen(false)] out TSource current)
             {
                 NotificationKind kind;
 
@@ -119,10 +118,10 @@ namespace System.Reactive.Linq.ObservableImpl
                 switch (kind)
                 {
                     case NotificationKind.OnNext:
-                        current = value;
+                        current = value!;
                         return true;
                     case NotificationKind.OnError:
-                        error.Throw();
+                        error!.Throw();
                         break;
                     case NotificationKind.OnCompleted:
                         break;

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

@@ -2,7 +2,7 @@
 // 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.Diagnostics.CodeAnalysis;
 
 namespace System.Reactive.Linq.ObservableImpl
 {
@@ -31,7 +31,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             private volatile NotificationKind _kind;
             private TSource _value;
-            private Exception _error;
+            private Exception? _error;
 
             public override void OnNext(TSource value)
             {
@@ -54,7 +54,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 _kind = NotificationKind.OnCompleted;  // Write last!
             }
 
-            public override bool TryMoveNext(out TSource current)
+            public override bool TryMoveNext([MaybeNullWhen(false)]out TSource current)
             {
                 //
                 // Notice the _kind field is marked volatile and read before the other fields.
@@ -68,7 +68,7 @@ namespace System.Reactive.Linq.ObservableImpl
                         current = _value;
                         return true;
                     case NotificationKind.OnError:
-                        _error.Throw();
+                        _error!.Throw();
                         break;
                     case NotificationKind.OnCompleted:
                         break;

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

@@ -2,8 +2,7 @@
 // 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.Diagnostics.CodeAnalysis;
 using System.Threading;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -15,10 +14,7 @@ namespace System.Reactive.Linq.ObservableImpl
         {
         }
 
-        protected override PushToPullSink<TSource, TSource> Run()
-        {
-            return new _();
-        }
+        protected override PushToPullSink<TSource, TSource> Run() => new _();
 
         private sealed class _ : PushToPullSink<TSource, TSource>
         {
@@ -33,8 +29,8 @@ namespace System.Reactive.Linq.ObservableImpl
 
             private bool _waiting;
             private NotificationKind _kind;
-            private TSource _value;
-            private Exception _error;
+            private TSource? _value;
+            private Exception? _error;
 
             public override void OnNext(TSource value)
             {
@@ -92,7 +88,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
             }
 
-            public override bool TryMoveNext(out TSource current)
+            public override bool TryMoveNext([MaybeNullWhen(false)] out TSource current)
             {
                 var done = false;
 
@@ -127,10 +123,10 @@ namespace System.Reactive.Linq.ObservableImpl
                 switch (_kind)
                 {
                     case NotificationKind.OnNext:
-                        current = _value;
+                        current = _value!;
                         return true;
                     case NotificationKind.OnError:
-                        _error.Throw();
+                        _error!.Throw();
                         break;
                     case NotificationKind.OnCompleted:
                         break;

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

@@ -2,10 +2,9 @@
 // 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;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Reactive.Disposables;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -33,13 +32,13 @@ namespace System.Reactive.Linq.ObservableImpl
 
     internal abstract class PushToPullSink<TSource, TResult> : IObserver<TSource>, IEnumerator<TResult>
     {
-        private IDisposable _upstream;
+        private IDisposable? _upstream;
 
         public abstract void OnNext(TSource value);
         public abstract void OnError(Exception error);
         public abstract void OnCompleted();
 
-        public abstract bool TryMoveNext(out TResult current);
+        public abstract bool TryMoveNext([MaybeNullWhen(false)] out TResult current);
 
         private bool _done;
 
@@ -60,6 +59,7 @@ namespace System.Reactive.Linq.ObservableImpl
             return false;
         }
 
+#nullable disable // NB: Matches the protocol around accessing Current only if MoveNext returns true.
         public TResult Current
         {
             get;
@@ -67,6 +67,7 @@ namespace System.Reactive.Linq.ObservableImpl
         }
 
         object IEnumerator.Current => Current;
+#nullable restore
 
         public void Reset()
         {