Merge pull request #1345 from dotnet/dev/bartde/rx_nullable_part9

Enable #nullable for easy query operator implementations.

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

@@ -2,6 +2,8 @@
 // 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 // TODO: Substitute for implementation that doesn't use DictionaryK, V>.
+
 using System.Collections;
 using System.Collections.Generic;
 using System.Linq;
@@ -9,7 +11,6 @@ using System.Linq;
 namespace System.Reactive
 {
     internal sealed class Lookup<K, E> : ILookup<K, E>
-        where K : notnull
     {
         private readonly Dictionary<K, List<E>> _dictionary;
 

Rx.NET/Source/src/System.Reactive/Linq/IQueryLanguage.cs

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information. 
 
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Reactive.Concurrency;
 using System.Reactive.Joins;
@@ -338,21 +339,21 @@ namespace System.Reactive.Linq
         IEnumerable<TResult> Collect<TSource, TResult>(IObservable<TSource> source, Func<TResult> getInitialCollector, Func<TResult, TSource, TResult> merge, Func<TResult, TResult> getNewCollector);
         TSource First<TSource>(IObservable<TSource> source);
         TSource First<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
-        TSource FirstOrDefault<TSource>(IObservable<TSource> source);
-        TSource FirstOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
+        [return: MaybeNull] TSource FirstOrDefault<TSource>(IObservable<TSource> source);
+        [return: MaybeNull] TSource FirstOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
         void ForEach<TSource>(IObservable<TSource> source, Action<TSource> onNext);
         void ForEach<TSource>(IObservable<TSource> source, Action<TSource, int> onNext);
         IEnumerator<TSource> GetEnumerator<TSource>(IObservable<TSource> source);
         TSource Last<TSource>(IObservable<TSource> source);
         TSource Last<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
-        TSource LastOrDefault<TSource>(IObservable<TSource> source);
-        TSource LastOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
+        [return: MaybeNull] TSource LastOrDefault<TSource>(IObservable<TSource> source);
+        [return: MaybeNull] TSource LastOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
         IEnumerable<TSource> Latest<TSource>(IObservable<TSource> source);
         IEnumerable<TSource> MostRecent<TSource>(IObservable<TSource> source, TSource initialValue);
         IEnumerable<TSource> Next<TSource>(IObservable<TSource> source);
         TSource Single<TSource>(IObservable<TSource> source);
-        TSource SingleOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
-        TSource SingleOrDefault<TSource>(IObservable<TSource> source);
+        [return: MaybeNull] TSource SingleOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
+        [return: MaybeNull] TSource SingleOrDefault<TSource>(IObservable<TSource> source);
         TSource Single<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
         TSource Wait<TSource>(IObservable<TSource> source);
 

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

@@ -3,6 +3,7 @@
 // See the LICENSE file in the project root for more information. 
 
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 
 namespace System.Reactive.Linq
 {
@@ -160,6 +161,7 @@ namespace System.Reactive.Linq
         /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
         /// <seealso cref="Observable.FirstOrDefaultAsync{TSource}(IObservable{TSource})"/>
         [Obsolete(Constants_Linq.UseAsync)]
+        [return: MaybeNull]
         public static TSource FirstOrDefault<TSource>(this IObservable<TSource> source)
         {
             if (source == null)
@@ -180,6 +182,7 @@ namespace System.Reactive.Linq
         /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="predicate"/> is null.</exception>
         /// <seealso cref="Observable.FirstOrDefaultAsync{TSource}(IObservable{TSource}, Func{TSource, bool})"/>
         [Obsolete(Constants_Linq.UseAsync)]
+        [return: MaybeNull]
         public static TSource FirstOrDefault<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             if (source == null)
@@ -331,6 +334,7 @@ namespace System.Reactive.Linq
         /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
         /// <seealso cref="Observable.LastOrDefaultAsync{TSource}(IObservable{TSource})"/>
         [Obsolete(Constants_Linq.UseAsync)]
+        [return: MaybeNull]
         public static TSource LastOrDefault<TSource>(this IObservable<TSource> source)
         {
             if (source == null)
@@ -351,6 +355,7 @@ namespace System.Reactive.Linq
         /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="predicate"/> is null.</exception>
         /// <seealso cref="Observable.LastOrDefaultAsync{TSource}(IObservable{TSource}, Func{TSource, bool})"/>
         [Obsolete(Constants_Linq.UseAsync)]
+        [return: MaybeNull]
         public static TSource LastOrDefault<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             if (source == null)
@@ -496,6 +501,7 @@ namespace System.Reactive.Linq
         /// <exception cref="InvalidOperationException">The source sequence contains more than one element.</exception>
         /// <seealso cref="Observable.SingleOrDefaultAsync{TSource}(IObservable{TSource})"/>
         [Obsolete(Constants_Linq.UseAsync)]
+        [return: MaybeNull]
         public static TSource SingleOrDefault<TSource>(this IObservable<TSource> source)
         {
             if (source == null)
@@ -517,6 +523,7 @@ namespace System.Reactive.Linq
         /// <exception cref="InvalidOperationException">The sequence contains more than one element that satisfies the condition in the predicate.</exception>
         /// <seealso cref="Observable.SingleOrDefaultAsync{TSource}(IObservable{TSource}, Func{TSource, bool})"/>
         [Obsolete(Constants_Linq.UseAsync)]
+        [return: MaybeNull]
         public static TSource SingleOrDefault<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             if (source == null)

Rx.NET/Source/src/System.Reactive/Linq/Observable/AddRef.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.Disposables;
 
 namespace System.Reactive.Linq.ObservableImpl

Rx.NET/Source/src/System.Reactive/Linq/Observable/Aggregate.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 sealed class Aggregate<TSource> : Producer<TSource, Aggregate<TSource>._>
@@ -24,7 +22,7 @@ namespace System.Reactive.Linq.ObservableImpl
         internal sealed class _ : IdentitySink<TSource>
         {
             private readonly Func<TSource, TSource, TSource> _accumulator;
-            private TSource _accumulation;
+            private TSource? _accumulation;
             private bool _hasAccumulation;
 
             public _(Func<TSource, TSource, TSource> accumulator, IObserver<TSource> observer)
@@ -44,7 +42,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     try
                     {
-                        _accumulation = _accumulator(_accumulation, value);
+                        _accumulation = _accumulator(_accumulation!, value);
                     }
                     catch (Exception exception)
                     {
@@ -75,7 +73,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
                 else
                 {
-                    var accumulation = _accumulation;
+                    var accumulation = _accumulation!;
                     _accumulation = default;
                     ForwardOnNext(accumulation);
                     ForwardOnCompleted();
@@ -104,7 +102,7 @@ namespace System.Reactive.Linq.ObservableImpl
         internal sealed class _ : Sink<TSource, TAccumulate>
         {
             private readonly Func<TAccumulate, TSource, TAccumulate> _accumulator;
-            private TAccumulate _accumulation;
+            private TAccumulate? _accumulation;
 
             public _(TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> accumulator, IObserver<TAccumulate> observer)
                 : base(observer)
@@ -117,7 +115,7 @@ namespace System.Reactive.Linq.ObservableImpl
             {
                 try
                 {
-                    _accumulation = _accumulator(_accumulation, value);
+                    _accumulation = _accumulator(_accumulation!, value);
                 }
                 catch (Exception exception)
                 {
@@ -134,7 +132,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public override void OnCompleted()
             {
-                var accumulation = _accumulation;
+                var accumulation = _accumulation!;
                 _accumulation = default;
                 ForwardOnNext(accumulation);
                 ForwardOnCompleted();
@@ -166,7 +164,7 @@ namespace System.Reactive.Linq.ObservableImpl
             private readonly Func<TAccumulate, TSource, TAccumulate> _accumulator;
             private readonly Func<TAccumulate, TResult> _resultSelector;
 
-            private TAccumulate _accumulation;
+            private TAccumulate? _accumulation;
 
             public _(Aggregate<TSource, TAccumulate, TResult> parent, IObserver<TResult> observer)
                 : base(observer)
@@ -181,25 +179,30 @@ namespace System.Reactive.Linq.ObservableImpl
             {
                 try
                 {
-                    _accumulation = _accumulator(_accumulation, value);
+                    _accumulation = _accumulator(_accumulation!, value);
                 }
                 catch (Exception exception)
                 {
+                    _accumulation = default;
                     ForwardOnError(exception);
                 }
             }
 
             public override void OnError(Exception error)
             {
+                _accumulation = default;
                 ForwardOnError(error);
             }
 
             public override void OnCompleted()
             {
+                var accumulation = _accumulation!;
+                _accumulation = default;
+
                 TResult result;
                 try
                 {
-                    result = _resultSelector(_accumulation);
+                    result = _resultSelector(accumulation);
                 }
                 catch (Exception exception)
                 {

Rx.NET/Source/src/System.Reactive/Linq/Observable/AutoConnect.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.Subjects;
 using System.Threading;
 
@@ -18,10 +16,10 @@ namespace System.Reactive.Linq.ObservableImpl
     {
         private readonly IConnectableObservable<T> _source;
         private readonly int _minObservers;
-        private readonly Action<IDisposable> _onConnect;
+        private readonly Action<IDisposable>? _onConnect;
         private int _count;
 
-        internal AutoConnect(IConnectableObservable<T> source, int minObservers, Action<IDisposable> onConnect)
+        internal AutoConnect(IConnectableObservable<T> source, int minObservers, Action<IDisposable>? onConnect)
         {
             _source = source;
             _minObservers = minObservers;

Rx.NET/Source/src/System.Reactive/Linq/Observable/Average.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 sealed class AverageDouble : Producer<double, AverageDouble._>

Rx.NET/Source/src/System.Reactive/Linq/Observable/Cast.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 sealed class Cast<TSource, TResult> : Producer<TResult, Cast<TSource, TResult>._> /* Could optimize further by deriving from Select<TResult> and providing Combine<TResult2>. We're not doing this (yet) for debuggability. */
@@ -31,7 +29,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 TResult result;
                 try
                 {
-                    result = (TResult)(object)value;
+                    result = (TResult)(object?)value;
                 }
                 catch (Exception exception)
                 {
@@ -39,7 +37,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     return;
                 }
 
-                ForwardOnNext(result);
+                ForwardOnNext(result!);
             }
         }
     }

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

@@ -2,9 +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 sealed class Defer<TValue> : Producer<TValue, Defer<TValue>._>, IEvaluatableObservable<TValue>

Rx.NET/Source/src/System.Reactive/Linq/Observable/Dematerialize.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 sealed class Dematerialize<TSource> : Producer<TSource, Dematerialize<TSource>._>
@@ -34,7 +32,7 @@ namespace System.Reactive.Linq.ObservableImpl
                         ForwardOnNext(value.Value);
                         break;
                     case NotificationKind.OnError:
-                        ForwardOnError(value.Exception);
+                        ForwardOnError(value.Exception!);
                         break;
                     case NotificationKind.OnCompleted:
                         ForwardOnCompleted();

Rx.NET/Source/src/System.Reactive/Linq/Observable/Distinct.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;
 
 namespace System.Reactive.Linq.ObservableImpl

Rx.NET/Source/src/System.Reactive/Linq/Observable/DistinctUntilChanged.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;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -30,7 +28,7 @@ namespace System.Reactive.Linq.ObservableImpl
             private readonly Func<TSource, TKey> _keySelector;
             private readonly IEqualityComparer<TKey> _comparer;
 
-            private TKey _currentKey;
+            private TKey? _currentKey;
             private bool _hasCurrentKey;
 
             public _(DistinctUntilChanged<TSource, TKey> parent, IObserver<TSource> observer)
@@ -58,7 +56,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     try
                     {
-                        comparerEquals = _comparer.Equals(_currentKey, key);
+                        comparerEquals = _comparer.Equals(_currentKey!, key);
                     }
                     catch (Exception exception)
                     {

Rx.NET/Source/src/System.Reactive/Linq/Observable/Do.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 Do<TSource>

Rx.NET/Source/src/System.Reactive/Linq/Observable/DoWhile.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;
 
 namespace System.Reactive.Linq.ObservableImpl

Rx.NET/Source/src/System.Reactive/Linq/Observable/ElementAt.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 sealed class ElementAt<TSource> : Producer<TSource, ElementAt<TSource>._>

Rx.NET/Source/src/System.Reactive/Linq/Observable/ElementAtOrDefault.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 sealed class ElementAtOrDefault<TSource> : Producer<TSource, ElementAtOrDefault<TSource>._>
@@ -44,7 +42,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public override void OnCompleted()
             {
-                ForwardOnNext(default);
+                ForwardOnNext(default!);
                 ForwardOnCompleted();
             }
         }

Rx.NET/Source/src/System.Reactive/Linq/Observable/FirstAsync.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 FirstAsync<TSource>

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

@@ -2,17 +2,15 @@
 // 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.Threading;
 
 namespace System.Reactive.Linq.ObservableImpl
 {
     internal abstract class BaseBlocking<T> : ManualResetEventSlim, IObserver<T>
     {
-        internal T _value;
+        internal T? _value;
         internal bool _hasValue;
-        internal Exception _error;
+        internal Exception? _error;
 
         internal BaseBlocking() { }
 

Rx.NET/Source/src/System.Reactive/Linq/Observable/FirstOrDefaultAsync.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 FirstOrDefaultAsync<TSource>
@@ -36,7 +34,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    ForwardOnNext(default);
+                    ForwardOnNext(default!);
                     ForwardOnCompleted();
                 }
             }
@@ -90,7 +88,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    ForwardOnNext(default);
+                    ForwardOnNext(default!);
                     ForwardOnCompleted();
                 }
             }

Rx.NET/Source/src/System.Reactive/Linq/Observable/For.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;
 
 namespace System.Reactive.Linq.ObservableImpl

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

@@ -2,9 +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 sealed class If<TResult> : Producer<TResult, If<TResult>._>, IEvaluatableObservable<TResult>

Rx.NET/Source/src/System.Reactive/Linq/Observable/LastAsync.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 LastAsync<TSource>
@@ -23,7 +21,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal sealed class _ : IdentitySink<TSource>
             {
-                private TSource _value;
+                private TSource? _value;
                 private bool _seenValue;
 
                 public _(IObserver<TSource> observer)
@@ -59,7 +57,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     }
                     else
                     {
-                        var value = _value;
+                        var value = _value!;
                         _value = default;
                         ForwardOnNext(value);
                         ForwardOnCompleted();
@@ -86,7 +84,7 @@ namespace System.Reactive.Linq.ObservableImpl
             internal sealed class _ : IdentitySink<TSource>
             {
                 private readonly Func<TSource, bool> _predicate;
-                private TSource _value;
+                private TSource? _value;
                 private bool _seenValue;
 
                 public _(Func<TSource, bool> predicate, IObserver<TSource> observer)
@@ -138,7 +136,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     }
                     else
                     {
-                        var value = _value;
+                        var value = _value!;
                         _value = default;
                         ForwardOnNext(value);
                         ForwardOnCompleted();

Rx.NET/Source/src/System.Reactive/Linq/Observable/LastOrDefaultAsync.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 LastOrDefaultAsync<TSource>
@@ -23,7 +21,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal sealed class _ : IdentitySink<TSource>
             {
-                private TSource _value;
+                private TSource? _value;
 
                 public _(IObserver<TSource> observer)
                     : base(observer)
@@ -44,7 +42,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    var value = _value;
+                    var value = _value!;
                     _value = default;
                     ForwardOnNext(value);
                     ForwardOnCompleted();
@@ -70,7 +68,7 @@ namespace System.Reactive.Linq.ObservableImpl
             internal sealed class _ : IdentitySink<TSource>
             {
                 private readonly Func<TSource, bool> _predicate;
-                private TSource _value;
+                private TSource? _value;
 
                 public _(Func<TSource, bool> predicate, IObserver<TSource> observer)
                     : base(observer)
@@ -107,7 +105,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    var value = _value;
+                    var value = _value!;
                     _value = default;
                     ForwardOnNext(value);
                     ForwardOnCompleted();

Rx.NET/Source/src/System.Reactive/Linq/Observable/Materialize.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 sealed class Materialize<TSource> : Producer<Notification<TSource>, Materialize<TSource>._>

Rx.NET/Source/src/System.Reactive/Linq/Observable/Max.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;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -37,7 +35,7 @@ namespace System.Reactive.Linq.ObservableImpl
         private sealed class NonNull : _
         {
             private bool _hasValue;
-            private TSource _lastValue;
+            private TSource? _lastValue;
 
             public NonNull(IComparer<TSource> comparer, IObserver<TSource> observer)
                 : base(comparer, observer)
@@ -51,7 +49,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     int comparison;
                     try
                     {
-                        comparison = _comparer.Compare(value, _lastValue);
+                        comparison = _comparer.Compare(value, _lastValue!);
                     }
                     catch (Exception ex)
                     {
@@ -86,7 +84,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
                 else
                 {
-                    ForwardOnNext(_lastValue);
+                    ForwardOnNext(_lastValue!);
                     ForwardOnCompleted();
                 }
             }
@@ -94,7 +92,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
         private sealed class Null : _
         {
-            private TSource _lastValue;
+            private TSource? _lastValue;
 
             public Null(IComparer<TSource> comparer, IObserver<TSource> observer)
                 : base(comparer, observer)
@@ -137,7 +135,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public override void OnCompleted()
             {
-                ForwardOnNext(_lastValue);
+                ForwardOnNext(_lastValue!);
                 ForwardOnCompleted();
             }
         }

Rx.NET/Source/src/System.Reactive/Linq/Observable/MaxBy.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;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -30,7 +28,7 @@ namespace System.Reactive.Linq.ObservableImpl
             private readonly Func<TSource, TKey> _keySelector;
             private readonly IComparer<TKey> _comparer;
             private bool _hasValue;
-            private TKey _lastKey;
+            private TKey? _lastKey;
             private List<TSource> _list;
 
             public _(MaxBy<TSource, TKey> parent, IObserver<IList<TSource>> observer)
@@ -51,8 +49,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
                 catch (Exception ex)
                 {
-                    _list = null;
-                    _lastKey = default;
+                    Cleanup();
                     ForwardOnError(ex);
                     return;
                 }
@@ -68,12 +65,11 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     try
                     {
-                        comparison = _comparer.Compare(key, _lastKey);
+                        comparison = _comparer.Compare(key, _lastKey!);
                     }
                     catch (Exception ex)
                     {
-                        _list = null;
-                        _lastKey = default;
+                        Cleanup();
                         ForwardOnError(ex);
                         return;
                     }
@@ -93,19 +89,23 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public override void OnError(Exception error)
             {
-                _lastKey = default;
-                _list = null;
+                Cleanup();
                 base.OnError(error);
             }
 
             public override void OnCompleted()
             {
                 var list = _list;
-                _list = null;
-                _lastKey = default;
+                Cleanup();
                 ForwardOnNext(list);
                 ForwardOnCompleted();
             }
+
+            private void Cleanup()
+            {
+                _list = null!;
+                _lastKey = default;
+            }
         }
     }
 }

Rx.NET/Source/src/System.Reactive/Linq/Observable/Min.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;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -37,7 +35,7 @@ namespace System.Reactive.Linq.ObservableImpl
         private sealed class NonNull : _
         {
             private bool _hasValue;
-            private TSource _lastValue;
+            private TSource? _lastValue;
 
             public NonNull(IComparer<TSource> comparer, IObserver<TSource> observer)
                 : base(comparer, observer)
@@ -51,7 +49,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     int comparison;
                     try
                     {
-                        comparison = _comparer.Compare(value, _lastValue);
+                        comparison = _comparer.Compare(value, _lastValue!);
                     }
                     catch (Exception ex)
                     {
@@ -91,7 +89,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
                 else
                 {
-                    ForwardOnNext(_lastValue);
+                    ForwardOnNext(_lastValue!);
                     ForwardOnCompleted();
                 }
             }
@@ -99,7 +97,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
         private sealed class Null : _
         {
-            private TSource _lastValue;
+            private TSource? _lastValue;
 
             public Null(IComparer<TSource> comparer, IObserver<TSource> observer)
                 : base(comparer, observer)
@@ -137,7 +135,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public override void OnCompleted()
             {
-                ForwardOnNext(_lastValue);
+                ForwardOnNext(_lastValue!);
                 ForwardOnCompleted();
             }
         }

Rx.NET/Source/src/System.Reactive/Linq/Observable/MinBy.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;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -30,7 +28,7 @@ namespace System.Reactive.Linq.ObservableImpl
             private readonly Func<TSource, TKey> _keySelector;
             private readonly IComparer<TKey> _comparer;
             private bool _hasValue;
-            private TKey _lastKey;
+            private TKey? _lastKey;
             private List<TSource> _list;
 
             public _(MinBy<TSource, TKey> parent, IObserver<IList<TSource>> observer)
@@ -51,8 +49,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
                 catch (Exception ex)
                 {
-                    _list = null;
-                    _lastKey = default;
+                    Cleanup();
                     ForwardOnError(ex);
                     return;
                 }
@@ -68,12 +65,11 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     try
                     {
-                        comparison = _comparer.Compare(key, _lastKey);
+                        comparison = _comparer.Compare(key, _lastKey!);
                     }
                     catch (Exception ex)
                     {
-                        _list = null;
-                        _lastKey = default;
+                        Cleanup();
                         ForwardOnError(ex);
                         return;
                     }
@@ -94,11 +90,16 @@ namespace System.Reactive.Linq.ObservableImpl
             public override void OnCompleted()
             {
                 var list = _list;
-                _list = null;
-                _lastKey = default;
+                Cleanup();
                 ForwardOnNext(list);
                 ForwardOnCompleted();
             }
+
+            private void Cleanup()
+            {
+                _list = null!;
+                _lastKey = default;
+            }
         }
     }
 }

Rx.NET/Source/src/System.Reactive/Linq/Observable/Scan.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 sealed class Scan<TSource, TAccumulate> : Producer<TAccumulate, Scan<TSource, TAccumulate>._>
@@ -70,7 +68,7 @@ namespace System.Reactive.Linq.ObservableImpl
         internal sealed class _ : IdentitySink<TSource>
         {
             private readonly Func<TSource, TSource, TSource> _accumulator;
-            private TSource _accumulation;
+            private TSource? _accumulation;
             private bool _hasAccumulation;
 
             public _(Func<TSource, TSource, TSource> accumulator, IObserver<TSource> observer)
@@ -85,7 +83,7 @@ namespace System.Reactive.Linq.ObservableImpl
                 {
                     if (_hasAccumulation)
                     {
-                        _accumulation = _accumulator(_accumulation, value);
+                        _accumulation = _accumulator(_accumulation!, value);
                     }
                     else
                     {

Rx.NET/Source/src/System.Reactive/Linq/Observable/Select.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 Select<TSource, TResult>

Rx.NET/Source/src/System.Reactive/Linq/Observable/SingleAsync.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 SingleAsync<TSource>
@@ -23,7 +21,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal sealed class _ : IdentitySink<TSource>
             {
-                private TSource _value;
+                private TSource? _value;
                 private bool _seenValue;
 
                 public _(IObserver<TSource> observer)
@@ -65,7 +63,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     }
                     else
                     {
-                        ForwardOnNext(_value);
+                        ForwardOnNext(_value!);
                         ForwardOnCompleted();
                     }
                 }
@@ -90,7 +88,7 @@ namespace System.Reactive.Linq.ObservableImpl
             internal sealed class _ : IdentitySink<TSource>
             {
                 private readonly Func<TSource, bool> _predicate;
-                private TSource _value;
+                private TSource? _value;
                 private bool _seenValue;
 
                 public _(Func<TSource, bool> predicate, IObserver<TSource> observer)
@@ -148,7 +146,7 @@ namespace System.Reactive.Linq.ObservableImpl
                     }
                     else
                     {
-                        ForwardOnNext(_value);
+                        ForwardOnNext(_value!);
                         ForwardOnCompleted();
                     }
                 }

Rx.NET/Source/src/System.Reactive/Linq/Observable/SingleOrDefaultAsync.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 SingleOrDefaultAsync<TSource>
@@ -23,7 +21,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
             internal sealed class _ : IdentitySink<TSource>
             {
-                private TSource _value;
+                private TSource? _value;
                 private bool _seenValue;
 
                 public _(IObserver<TSource> observer)
@@ -52,7 +50,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    ForwardOnNext(_value);
+                    ForwardOnNext(_value!);
                     ForwardOnCompleted();
                 }
             }
@@ -76,7 +74,7 @@ namespace System.Reactive.Linq.ObservableImpl
             internal sealed class _ : IdentitySink<TSource>
             {
                 private readonly Func<TSource, bool> _predicate;
-                private TSource _value;
+                private TSource? _value;
                 private bool _seenValue;
 
                 public _(Func<TSource, bool> predicate, IObserver<TSource> observer)
@@ -121,7 +119,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    ForwardOnNext(_value);
+                    ForwardOnNext(_value!);
                     ForwardOnCompleted();
                 }
             }

Rx.NET/Source/src/System.Reactive/Linq/Observable/ToArray.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;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -38,17 +36,22 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public override void OnError(Exception error)
             {
-                _list = null;
+                Cleanup();
                 base.OnError(error);
             }
 
             public override void OnCompleted()
             {
                 var list = _list;
-                _list = null;
+                Cleanup();
                 ForwardOnNext(list.ToArray());
                 ForwardOnCompleted();
             }
+
+            private void Cleanup()
+            {
+                _list = null!;
+            }
         }
     }
 }

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

@@ -2,13 +2,12 @@
 // 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;
 
 namespace System.Reactive.Linq.ObservableImpl
 {
     internal sealed class ToDictionary<TSource, TKey, TElement> : Producer<IDictionary<TKey, TElement>, ToDictionary<TSource, TKey, TElement>._>
+        where TKey : notnull
     {
         private readonly IObservable<TSource> _source;
         private readonly Func<TSource, TKey> _keySelector;
@@ -49,24 +48,29 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
                 catch (Exception ex)
                 {
-                    _dictionary = null;
+                    Cleanup();
                     ForwardOnError(ex);
                 }
             }
 
             public override void OnError(Exception error)
             {
-                _dictionary = null;
+                Cleanup();
                 ForwardOnError(error);
             }
 
             public override void OnCompleted()
             {
                 var dictionary = _dictionary;
-                _dictionary = null;
+                Cleanup();
                 ForwardOnNext(dictionary);
                 ForwardOnCompleted();
             }
+
+            private void Cleanup()
+            {
+                _dictionary = null!;
+            }
         }
     }
 }

Rx.NET/Source/src/System.Reactive/Linq/Observable/ToList.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;
 
 namespace System.Reactive.Linq.ObservableImpl
@@ -38,17 +36,22 @@ namespace System.Reactive.Linq.ObservableImpl
 
             public override void OnError(Exception error)
             {
-                _list = null;
+                Cleanup();
                 ForwardOnError(error);
             }
 
             public override void OnCompleted()
             {
                 var list = _list;
-                _list = null;
+                Cleanup();
                 ForwardOnNext(list);
                 ForwardOnCompleted();
             }
+
+            private void Cleanup()
+            {
+                _list = null!;
+            }
         }
     }
 }

Rx.NET/Source/src/System.Reactive/Linq/Observable/ToLookup.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.Linq;
 
@@ -50,24 +48,29 @@ namespace System.Reactive.Linq.ObservableImpl
                 }
                 catch (Exception ex)
                 {
-                    _lookup = null;
+                    Cleanup();
                     ForwardOnError(ex);
                 }
             }
 
             public override void OnError(Exception error)
             {
-                _lookup = null;
+                Cleanup();
                 ForwardOnError(error);
             }
 
             public override void OnCompleted()
             {
                 var lookup = _lookup;
-                _lookup = null;
+                Cleanup();
                 ForwardOnNext(lookup);
                 ForwardOnCompleted();
             }
+
+            private void Cleanup()
+            {
+                _lookup = null!;
+            }
         }
     }
 }

Rx.NET/Source/src/System.Reactive/Linq/Observable/Where.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 Where<TSource>

Rx.NET/Source/src/System.Reactive/Linq/Observable/While.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;
 
 namespace System.Reactive.Linq.ObservableImpl

Rx.NET/Source/src/System.Reactive/Linq/QueryLanguage.Blocking.cs

@@ -7,6 +7,8 @@ using System.Threading;
 
 namespace System.Reactive.Linq
 {
+    using System.Diagnostics.CodeAnalysis;
+
     using ObservableImpl;
 
     internal partial class QueryLanguage
@@ -43,7 +45,7 @@ namespace System.Reactive.Linq
 
         public virtual TSource First<TSource>(IObservable<TSource> source)
         {
-            return FirstOrDefaultInternal(source, true);
+            return FirstOrDefaultInternal(source, true)!;
         }
 
         public virtual TSource First<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
@@ -55,16 +57,19 @@ namespace System.Reactive.Linq
 
         #region FirstOrDefault
 
+        [return: MaybeNull]
         public virtual TSource FirstOrDefault<TSource>(IObservable<TSource> source)
         {
             return FirstOrDefaultInternal(source, false);
         }
 
+        [return: MaybeNull]
         public virtual TSource FirstOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             return FirstOrDefault(Where(source, predicate));
         }
 
+        [return: MaybeNull]
         private static TSource FirstOrDefaultInternal<TSource>(IObservable<TSource> source, bool throwOnEmpty)
         {
             using (var consumer = new FirstBlocking<TSource>())
@@ -80,6 +85,7 @@ namespace System.Reactive.Linq
                 {
                     throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
                 }
+
                 return consumer._value;
             }
         }
@@ -134,7 +140,7 @@ namespace System.Reactive.Linq
 
         public virtual TSource Last<TSource>(IObservable<TSource> source)
         {
-            return LastOrDefaultInternal(source, true);
+            return LastOrDefaultInternal(source, true)!;
         }
 
         public virtual TSource Last<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
@@ -146,16 +152,19 @@ namespace System.Reactive.Linq
 
         #region LastOrDefault
 
+        [return: MaybeNull]
         public virtual TSource LastOrDefault<TSource>(IObservable<TSource> source)
         {
             return LastOrDefaultInternal(source, false);
         }
 
+        [return: MaybeNull]
         public virtual TSource LastOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             return LastOrDefault(Where(source, predicate));
         }
 
+        [return: MaybeNull]
         private static TSource LastOrDefaultInternal<TSource>(IObservable<TSource> source, bool throwOnEmpty)
         {
             using (var consumer = new LastBlocking<TSource>())
@@ -172,6 +181,7 @@ namespace System.Reactive.Linq
                 {
                     throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
                 }
+
                 return consumer._value;
             }
         }
@@ -209,7 +219,7 @@ namespace System.Reactive.Linq
 
         public virtual TSource Single<TSource>(IObservable<TSource> source)
         {
-            return SingleOrDefaultInternal(source, true);
+            return SingleOrDefaultInternal(source, true)!;
         }
 
         public virtual TSource Single<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
@@ -221,16 +231,19 @@ namespace System.Reactive.Linq
 
         #region SingleOrDefault
 
+        [return: MaybeNull]
         public virtual TSource SingleOrDefault<TSource>(IObservable<TSource> source)
         {
             return SingleOrDefaultInternal(source, false);
         }
 
+        [return: MaybeNull]
         public virtual TSource SingleOrDefault<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             return SingleOrDefault(Where(source, predicate));
         }
 
+        [return: MaybeNull]
         private static TSource SingleOrDefaultInternal<TSource>(IObservable<TSource> source, bool throwOnEmpty)
         {
             var value = default(TSource);
@@ -281,7 +294,7 @@ namespace System.Reactive.Linq
                 throw new InvalidOperationException(Strings_Linq.NO_ELEMENTS);
             }
 
-            return value!;
+            return value;
         }
 
         #endregion
@@ -290,7 +303,7 @@ namespace System.Reactive.Linq
 
         public virtual TSource Wait<TSource>(IObservable<TSource> source)
         {
-            return LastOrDefaultInternal(source, true);
+            return LastOrDefaultInternal(source, true)!;
         }
 
         #endregion

Rx.NET/Source/tests/Tests.System.Reactive.ApiApprovals/Api/ApiApprovalTests.Core.verified.cs

@@ -1027,8 +1027,10 @@ namespace System.Reactive.Linq
         public static System.IObservable<TSource> FirstAsync<TSource>(this System.IObservable<TSource> source) { }
         public static System.IObservable<TSource> FirstAsync<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
         [System.Obsolete(@"This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block.")]
+        [return: System.Diagnostics.CodeAnalysis.MaybeNull]
         public static TSource FirstOrDefault<TSource>(this System.IObservable<TSource> source) { }
         [System.Obsolete(@"This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block.")]
+        [return: System.Diagnostics.CodeAnalysis.MaybeNull]
         public static TSource FirstOrDefault<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
         public static System.IObservable<TSource> FirstOrDefaultAsync<TSource>(this System.IObservable<TSource> source) { }
         public static System.IObservable<TSource> FirstOrDefaultAsync<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
@@ -1180,8 +1182,10 @@ namespace System.Reactive.Linq
         public static System.IObservable<TSource> LastAsync<TSource>(this System.IObservable<TSource> source) { }
         public static System.IObservable<TSource> LastAsync<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
         [System.Obsolete(@"This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block.")]
+        [return: System.Diagnostics.CodeAnalysis.MaybeNull]
         public static TSource LastOrDefault<TSource>(this System.IObservable<TSource> source) { }
         [System.Obsolete(@"This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block.")]
+        [return: System.Diagnostics.CodeAnalysis.MaybeNull]
         public static TSource LastOrDefault<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
         public static System.IObservable<TSource> LastOrDefaultAsync<TSource>(this System.IObservable<TSource> source) { }
         public static System.IObservable<TSource> LastOrDefaultAsync<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
@@ -1347,8 +1351,10 @@ namespace System.Reactive.Linq
         public static System.IObservable<TSource> SingleAsync<TSource>(this System.IObservable<TSource> source) { }
         public static System.IObservable<TSource> SingleAsync<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
         [System.Obsolete(@"This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block.")]
+        [return: System.Diagnostics.CodeAnalysis.MaybeNull]
         public static TSource SingleOrDefault<TSource>(this System.IObservable<TSource> source) { }
         [System.Obsolete(@"This blocking operation is no longer supported. Instead, use the async version in combination with C# and Visual Basic async/await support. In case you need a blocking operation, use Wait or convert the resulting observable sequence to a Task object and block.")]
+        [return: System.Diagnostics.CodeAnalysis.MaybeNull]
         public static TSource SingleOrDefault<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }
         public static System.IObservable<TSource> SingleOrDefaultAsync<TSource>(this System.IObservable<TSource> source) { }
         public static System.IObservable<TSource> SingleOrDefaultAsync<TSource>(this System.IObservable<TSource> source, System.Func<TSource, bool> predicate) { }