Merge pull request #1338 from dotnet/dev/bartde/rx_nullable_step2

Nullable fixes for internal types.

Rx.NET/Source/Directory.build.targets

@@ -7,19 +7,19 @@
   </PropertyGroup>
     
   <PropertyGroup Condition="'$(TargetFramework)' == 'net472'">
-    <DefineConstants>$(DefineConstants);HAS_WINFORMS;HAS_WPF;HAS_WINRT;HAS_DISPATCHER;HAS_REMOTING;DESKTOPCLR</DefineConstants>
+    <DefineConstants>$(DefineConstants);HAS_WINFORMS;HAS_WPF;HAS_WINRT;HAS_DISPATCHER;HAS_REMOTING;DESKTOPCLR;NO_NULLABLE_ATTRIBUTES</DefineConstants>
   </PropertyGroup>
   <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
     <TargetPlatformVersion>10.0.19041.0</TargetPlatformVersion>
     <TargetPlatformMinVersion>10.0.15063.0</TargetPlatformMinVersion>
-    <DefineConstants>$(DefineConstants);HAS_WINRT;NO_CODE_COVERAGE_ATTRIBUTE;NO_SERIALIZABLE;CRIPPLED_REFLECTION;NO_THREAD;NO_TRACE;WINDOWS</DefineConstants>
+    <DefineConstants>$(DefineConstants);HAS_WINRT;NO_CODE_COVERAGE_ATTRIBUTE;NO_SERIALIZABLE;CRIPPLED_REFLECTION;NO_THREAD;NO_TRACE;WINDOWS;NO_NULLABLE_ATTRIBUTES</DefineConstants>
   </PropertyGroup>
   <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0.16299'">
-    <DefineConstants>$(DefineConstants);HAS_WINRT;WINDOWS;HAS_OS_XAML;LEGACY_WINRT</DefineConstants>
+    <DefineConstants>$(DefineConstants);HAS_WINRT;WINDOWS;HAS_OS_XAML;LEGACY_WINRT;NO_NULLABLE_ATTRIBUTES</DefineConstants>
     <TargetPlatformVersion>10.0.19041.0</TargetPlatformVersion>
   </PropertyGroup>
   <PropertyGroup Condition="'$(TargetFramework)' == 'netstandard2.0' or '$(TargetFramework)' == 'netcoreapp2.1'">
-    <DefineConstants>$(DefineConstants);HAS_WINRT</DefineConstants>
+    <DefineConstants>$(DefineConstants);HAS_WINRT;NO_NULLABLE_ATTRIBUTES</DefineConstants>
   </PropertyGroup>
   <PropertyGroup Condition="'$(TargetFramework)' == 'netcoreapp3.1'">
     <DefineConstants>$(DefineConstants);HAS_WINRT;HAS_WINFORMS;HAS_WPF;HAS_DISPATCHER;DESKTOPCLR</DefineConstants>

Rx.NET/Source/src/System.Reactive/Concurrency/AsyncLock.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.Concurrency
@@ -16,7 +14,7 @@ namespace System.Reactive.Concurrency
         private bool _isAcquired;
         private bool _hasFaulted;
         private readonly object _guard = new object();
-        private Queue<(Action<Delegate, object> action, Delegate @delegate, object state)> _queue;
+        private Queue<(Action<Delegate, object?> action, Delegate @delegate, object? state)>? _queue;
 
         /// <summary>
         /// Queues the action for execution. If the caller acquires the lock and becomes the owner,
@@ -32,7 +30,7 @@ namespace System.Reactive.Concurrency
                 throw new ArgumentNullException(nameof(action));
             }
 
-            Wait(action, static closureAction => closureAction());
+            Wait(action, static closureAction => closureAction!());
         }
 
         /// <summary>
@@ -52,10 +50,10 @@ namespace System.Reactive.Concurrency
                 throw new ArgumentNullException(nameof(action));
             }
 
-            Wait(state, action, static (actionObject, stateObject) => ((Action<TState>)actionObject)((TState)stateObject));
+            Wait(state, action, static (actionObject, stateObject) => ((Action<TState>)actionObject)((TState)stateObject!));
         }
 
-        private void Wait(object state, Delegate @delegate, Action<Delegate, object> action)
+        private void Wait(object? state, Delegate @delegate, Action<Delegate, object?> action)
         {
             // allow one thread to update the state
             lock (_guard)
@@ -74,7 +72,7 @@ namespace System.Reactive.Concurrency
                     var q = _queue;
                     if (q == null)
                     {
-                        q = new Queue<(Action<Delegate, object> action, Delegate @delegate, object state)>();
+                        q = new Queue<(Action<Delegate, object?> action, Delegate @delegate, object? state)>();
                         _queue = q;
                     }
                     // enqueue the work

Rx.NET/Source/src/System.Reactive/Diagnostics/CodeAnalysis/NullableAttributes.cs

@@ -0,0 +1,148 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#if NO_NULLABLE_ATTRIBUTES
+
+namespace System.Diagnostics.CodeAnalysis
+{
+    /// <summary>Specifies that null is allowed as an input even if the corresponding type disallows it.</summary>
+    [AttributeUsage(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property, Inherited = false)]
+    internal sealed class AllowNullAttribute : Attribute
+    { }
+
+    /// <summary>Specifies that null is disallowed as an input even if the corresponding type allows it.</summary>
+    [AttributeUsage(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property, Inherited = false)]
+    internal sealed class DisallowNullAttribute : Attribute
+    { }
+
+    /// <summary>Specifies that an output may be null even if the corresponding type disallows it.</summary>
+    [AttributeUsage(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property | AttributeTargets.ReturnValue, Inherited = false)]
+    internal sealed class MaybeNullAttribute : Attribute
+    { }
+
+    /// <summary>Specifies that an output will not be null even if the corresponding type allows it. Specifies that an input argument was not null when the call returns.</summary>
+    [AttributeUsage(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property | AttributeTargets.ReturnValue, Inherited = false)]
+    internal sealed class NotNullAttribute : Attribute
+    { }
+
+    /// <summary>Specifies that when a method returns <see cref="ReturnValue"/>, the parameter may be null even if the corresponding type disallows it.</summary>
+    [AttributeUsage(AttributeTargets.Parameter, Inherited = false)]
+    internal sealed class MaybeNullWhenAttribute : Attribute
+    {
+        /// <summary>Initializes the attribute with the specified return value condition.</summary>
+        /// <param name="returnValue">
+        /// The return value condition. If the method returns this value, the associated parameter may be null.
+        /// </param>
+        public MaybeNullWhenAttribute(bool returnValue) => ReturnValue = returnValue;
+
+        /// <summary>Gets the return value condition.</summary>
+        public bool ReturnValue { get; }
+    }
+
+    /// <summary>Specifies that when a method returns <see cref="ReturnValue"/>, the parameter will not be null even if the corresponding type allows it.</summary>
+    [AttributeUsage(AttributeTargets.Parameter, Inherited = false)]
+    internal sealed class NotNullWhenAttribute : Attribute
+    {
+        /// <summary>Initializes the attribute with the specified return value condition.</summary>
+        /// <param name="returnValue">
+        /// The return value condition. If the method returns this value, the associated parameter will not be null.
+        /// </param>
+        public NotNullWhenAttribute(bool returnValue) => ReturnValue = returnValue;
+
+        /// <summary>Gets the return value condition.</summary>
+        public bool ReturnValue { get; }
+    }
+
+    /// <summary>Specifies that the output will be non-null if the named parameter is non-null.</summary>
+    [AttributeUsage(AttributeTargets.Parameter | AttributeTargets.Property | AttributeTargets.ReturnValue, AllowMultiple = true, Inherited = false)]
+    internal sealed class NotNullIfNotNullAttribute : Attribute
+    {
+        /// <summary>Initializes the attribute with the associated parameter name.</summary>
+        /// <param name="parameterName">
+        /// The associated parameter name.  The output will be non-null if the argument to the parameter specified is non-null.
+        /// </param>
+        public NotNullIfNotNullAttribute(string parameterName) => ParameterName = parameterName;
+
+        /// <summary>Gets the associated parameter name.</summary>
+        public string ParameterName { get; }
+    }
+
+    /// <summary>Applied to a method that will never return under any circumstance.</summary>
+    [AttributeUsage(AttributeTargets.Method, Inherited = false)]
+    internal sealed class DoesNotReturnAttribute : Attribute
+    { }
+
+    /// <summary>Specifies that the method will not return if the associated Boolean parameter is passed the specified value.</summary>
+    [AttributeUsage(AttributeTargets.Parameter, Inherited = false)]
+    internal sealed class DoesNotReturnIfAttribute : Attribute
+    {
+        /// <summary>Initializes the attribute with the specified parameter value.</summary>
+        /// <param name="parameterValue">
+        /// The condition parameter value. Code after the method will be considered unreachable by diagnostics if the argument to
+        /// the associated parameter matches this value.
+        /// </param>
+        public DoesNotReturnIfAttribute(bool parameterValue) => ParameterValue = parameterValue;
+
+        /// <summary>Gets the condition parameter value.</summary>
+        public bool ParameterValue { get; }
+    }
+
+    /// <summary>Specifies that the method or property will ensure that the listed field and property members have not-null values.</summary>
+    [AttributeUsage(AttributeTargets.Method | AttributeTargets.Property, Inherited = false, AllowMultiple = true)]
+    internal sealed class MemberNotNullAttribute : Attribute
+    {
+        /// <summary>Initializes the attribute with a field or property member.</summary>
+        /// <param name="member">
+        /// The field or property member that is promised to be not-null.
+        /// </param>
+        public MemberNotNullAttribute(string member) => Members = new[] { member };
+
+        /// <summary>Initializes the attribute with the list of field and property members.</summary>
+        /// <param name="members">
+        /// The list of field and property members that are promised to be not-null.
+        /// </param>
+        public MemberNotNullAttribute(params string[] members) => Members = members;
+
+        /// <summary>Gets field or property member names.</summary>
+        public string[] Members { get; }
+    }
+
+    /// <summary>Specifies that the method or property will ensure that the listed field and property members have not-null values when returning with the specified return value condition.</summary>
+    [AttributeUsage(AttributeTargets.Method | AttributeTargets.Property, Inherited = false, AllowMultiple = true)]
+    internal sealed class MemberNotNullWhenAttribute : Attribute
+    {
+        /// <summary>Initializes the attribute with the specified return value condition and a field or property member.</summary>
+        /// <param name="returnValue">
+        /// The return value condition. If the method returns this value, the associated parameter will not be null.
+        /// </param>
+        /// <param name="member">
+        /// The field or property member that is promised to be not-null.
+        /// </param>
+        public MemberNotNullWhenAttribute(bool returnValue, string member)
+        {
+            ReturnValue = returnValue;
+            Members = new[] { member };
+        }
+
+        /// <summary>Initializes the attribute with the specified return value condition and list of field and property members.</summary>
+        /// <param name="returnValue">
+        /// The return value condition. If the method returns this value, the associated parameter will not be null.
+        /// </param>
+        /// <param name="members">
+        /// The list of field and property members that are promised to be not-null.
+        /// </param>
+        public MemberNotNullWhenAttribute(bool returnValue, params string[] members)
+        {
+            ReturnValue = returnValue;
+            Members = members;
+        }
+
+        /// <summary>Gets the return value condition.</summary>
+        public bool ReturnValue { get; }
+
+        /// <summary>Gets field or property member names.</summary>
+        public string[] Members { get; }
+    }
+}
+
+#endif

Rx.NET/Source/src/System.Reactive/Internal/AnonymousEnumerable.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;
 using System.Collections.Generic;
 

Rx.NET/Source/src/System.Reactive/Internal/AsyncLockObserver.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;
 
 namespace System.Reactive

Rx.NET/Source/src/System.Reactive/Internal/AutoDetachObserver.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
@@ -12,7 +10,7 @@ namespace System.Reactive
     {
         private readonly IObserver<T> _observer;
 
-        private IDisposable _disposable;
+        private IDisposable? _disposable;
 
         public AutoDetachObserver(IObserver<T> observer)
         {

Rx.NET/Source/src/System.Reactive/Internal/BinaryObserver.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
 {
     internal sealed class BinaryObserver<TLeft, TRight> : IObserver<Either<Notification<TLeft>, Notification<TRight>>>

Rx.NET/Source/src/System.Reactive/Internal/CheckedObserver.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.Threading;
 
 namespace System.Reactive

Rx.NET/Source/src/System.Reactive/Internal/ConcatSink.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
@@ -15,7 +13,7 @@ namespace System.Reactive
         {
         }
 
-        protected override IEnumerable<IObservable<TSource>> Extract(IObservable<TSource> source) => (source as IConcatenatable<TSource>)?.GetSources();
+        protected override IEnumerable<IObservable<TSource>>? Extract(IObservable<TSource> source) => (source as IConcatenatable<TSource>)?.GetSources();
 
         public override void OnCompleted() => Recurse();
     }

Rx.NET/Source/src/System.Reactive/Internal/CurrentPlatformEnlightenmentProvider.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
-
 //
 // WARNING: The full namespace-qualified type name should stay the same for the discovery in System.Reactive.Core to work!
 //
@@ -27,7 +25,7 @@ namespace System.Reactive.PlatformServices
         /// <typeparam name="T">Service type.</typeparam>
         /// <param name="args">Optional set of arguments.</param>
         /// <returns>Service instance or <c>null</c> if not found.</returns>
-        public virtual T GetService<T>(object[] args) where T : class
+        public virtual T? GetService<T>(object[] args) where T : class
         {
             var t = typeof(T);
 
@@ -81,7 +79,7 @@ namespace System.Reactive.PlatformServices
                 {
                     var ifType = t.GetTypeInfo();
 
-                    var asm = new AssemblyName(ifType.Assembly.FullName)
+                    var asm = new AssemblyName(ifType.Assembly.FullName!)
                     {
                         Name = "System.Reactive"
                     };
@@ -91,7 +89,7 @@ namespace System.Reactive.PlatformServices
                     var dbg = Type.GetType(name, false);
                     if (dbg != null)
                     {
-                        return (T)Activator.CreateInstance(dbg);
+                        return (T?)Activator.CreateInstance(dbg);
                     }
                 }
             }

Rx.NET/Source/src/System.Reactive/Internal/Either.Generic.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.Globalization;
 
@@ -35,7 +33,7 @@ namespace System.Reactive
 
             public override void Switch(Action<TLeft> caseLeft, Action<TRight> caseRight) => caseLeft(Value);
 
-            public bool Equals(Left other)
+            public bool Equals(Left? other)
             {
                 if (other == this)
                 {
@@ -50,9 +48,9 @@ namespace System.Reactive
                 return EqualityComparer<TLeft>.Default.Equals(Value, other.Value);
             }
 
-            public override bool Equals(object obj) => Equals(obj as Left);
+            public override bool Equals(object? obj) => Equals(obj as Left);
 
-            public override int GetHashCode() => EqualityComparer<TLeft>.Default.GetHashCode(Value);
+            public override int GetHashCode() => Value?.GetHashCode() ?? 0;
 
             public override string ToString()
             {
@@ -73,7 +71,7 @@ namespace System.Reactive
 
             public override void Switch(Action<TLeft> caseLeft, Action<TRight> caseRight) => caseRight(Value);
 
-            public bool Equals(Right other)
+            public bool Equals(Right? other)
             {
                 if (other == this)
                 {
@@ -88,9 +86,9 @@ namespace System.Reactive
                 return EqualityComparer<TRight>.Default.Equals(Value, other.Value);
             }
 
-            public override bool Equals(object obj) => Equals(obj as Right);
+            public override bool Equals(object? obj) => Equals(obj as Right);
 
-            public override int GetHashCode() => EqualityComparer<TRight>.Default.GetHashCode(Value);
+            public override int GetHashCode() => Value?.GetHashCode() ?? 0;
 
             public override string ToString() => string.Format(CultureInfo.CurrentCulture, "Right({0})", Value);
         }

Rx.NET/Source/src/System.Reactive/Internal/ExceptionHelper.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.Threading;
 
@@ -29,7 +27,7 @@ namespace System.Reactive
         /// <param name="field">The target field to try to set atomically.</param>
         /// <param name="ex">The exception to set, not null (not verified).</param>
         /// <returns>True if the operation succeeded, false if the target was not null.</returns>
-        public static bool TrySetException(ref Exception field, Exception ex)
+        public static bool TrySetException(ref Exception? field, Exception ex)
         {
             return Interlocked.CompareExchange(ref field, ex, null) == null;
         }

Rx.NET/Source/src/System.Reactive/Internal/ExceptionServices.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.ComponentModel;
 using System.Reactive.PlatformServices;
 
@@ -15,7 +13,7 @@ namespace System.Reactive
 
         public static void Throw(this Exception exception) => Services.Value.Rethrow(exception);
 
-        public static void ThrowIfNotNull(this Exception exception)
+        public static void ThrowIfNotNull(this Exception? exception)
         {
             if (exception != null)
             {

Rx.NET/Source/src/System.Reactive/Internal/ExceptionServicesImpl.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.Runtime.ExceptionServices;
 
 namespace System.Reactive.PlatformServices
@@ -17,4 +15,4 @@ namespace System.Reactive.PlatformServices
     {
         public void Rethrow(Exception exception) => ExceptionDispatchInfo.Capture(exception).Throw();
     }
-}
+}

Rx.NET/Source/src/System.Reactive/Internal/HalfSerializer.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.Diagnostics;
 using System.Threading;
 
@@ -68,7 +66,7 @@ namespace System.Reactive
         /// <param name="ex">The exception to signal sooner or later.</param>
         /// <param name="wip">Indicates there is an emission going on currently.</param>
         /// <param name="error">The field containing an error or terminal indicator.</param>
-        public static void ForwardOnError<T>(ISink<T> sink, Exception ex, ref int wip, ref Exception error)
+        public static void ForwardOnError<T>(ISink<T> sink, Exception ex, ref int wip, ref Exception? error)
         {
             if (ExceptionHelper.TrySetException(ref error, ex))
             {
@@ -92,7 +90,7 @@ namespace System.Reactive
         /// <param name="sink">The observer to signal events in a serialized fashion.</param>
         /// <param name="wip">Indicates there is an emission going on currently.</param>
         /// <param name="error">The field containing an error or terminal indicator.</param>
-        public static void ForwardOnCompleted<T>(ISink<T> sink, ref int wip, ref Exception error)
+        public static void ForwardOnCompleted<T>(ISink<T> sink, ref int wip, ref Exception? error)
         {
             if (ExceptionHelper.TrySetException(ref error, ExceptionHelper.Terminated))
             {

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

@@ -18,7 +18,7 @@ namespace System.Reactive
             };
         }
 
-        public static IObservable<T> Unpack<T>(IObservable<T> source)
+        public static IObservable<T>? Unpack<T>(IObservable<T>? source)
         {
             bool hasOpt;
 

Rx.NET/Source/src/System.Reactive/Internal/HostLifecycleNotifications.Windows.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
-
 #if WINDOWS
 using Windows.ApplicationModel;
 using Windows.ApplicationModel.Core;
@@ -44,4 +42,4 @@ namespace System.Reactive.PlatformServices
         }
     }
 }
-#endif
+#endif

Rx.NET/Source/src/System.Reactive/Internal/HostLifecycleService.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.ComponentModel;
 using System.Threading;
 
@@ -15,19 +13,19 @@ namespace System.Reactive.PlatformServices
     [EditorBrowsable(EditorBrowsableState.Never)]
     public static class HostLifecycleService
     {
-        private static readonly Lazy<IHostLifecycleNotifications> Notifications = new Lazy<IHostLifecycleNotifications>(InitializeNotifications);
+        private static readonly Lazy<IHostLifecycleNotifications?> Notifications = new Lazy<IHostLifecycleNotifications?>(InitializeNotifications);
 
         private static int _refCount;
 
         /// <summary>
         /// Event that gets raised when the host suspends the application.
         /// </summary>
-        public static event EventHandler<HostSuspendingEventArgs> Suspending;
+        public static event EventHandler<HostSuspendingEventArgs>? Suspending;
 
         /// <summary>
         /// Event that gets raised when the host resumes the application.
         /// </summary>
-        public static event EventHandler<HostResumingEventArgs> Resuming;
+        public static event EventHandler<HostResumingEventArgs>? Resuming;
 
         /// <summary>
         /// Adds a reference to the host lifecycle manager, causing it to be sending notifications.
@@ -62,17 +60,17 @@ namespace System.Reactive.PlatformServices
             }
         }
 
-        private static void OnSuspending(object sender, HostSuspendingEventArgs e)
+        private static void OnSuspending(object? sender, HostSuspendingEventArgs e)
         {
             Suspending?.Invoke(sender, e);
         }
 
-        private static void OnResuming(object sender, HostResumingEventArgs e)
+        private static void OnResuming(object? sender, HostResumingEventArgs e)
         {
             Resuming?.Invoke(sender, e);
         }
 
-        private static IHostLifecycleNotifications InitializeNotifications()
+        private static IHostLifecycleNotifications? InitializeNotifications()
         {
 #pragma warning disable CS0618 // Type or member is obsolete
             return PlatformEnlightenmentProvider.Current.GetService<IHostLifecycleNotifications>();

Rx.NET/Source/src/System.Reactive/Internal/ImmutableList.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
 {
     internal sealed class ImmutableList<T>

Rx.NET/Source/src/System.Reactive/Internal/Lookup.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;
 using System.Collections.Generic;
 using System.Linq;
@@ -11,6 +9,7 @@ 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/Internal/Observers.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
 {
     internal sealed class NopObserver<T> : IObserver<T>

Rx.NET/Source/src/System.Reactive/Internal/PlatformEnlightenmentProvider.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.ComponentModel;
 
 namespace System.Reactive.PlatformServices
@@ -24,7 +22,7 @@ namespace System.Reactive.PlatformServices
         /// <typeparam name="T">Service type.</typeparam>
         /// <param name="args">Optional set of arguments.</param>
         /// <returns>Service instance or null if not found.</returns>
-        T GetService<T>(params object[] args) where T : class;
+        T? GetService<T>(params object[] args) where T : class;
     }
 
     /// <summary>

Rx.NET/Source/src/System.Reactive/Internal/PriorityQueue.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

Rx.NET/Source/src/System.Reactive/Internal/Producer.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;
 
@@ -42,7 +40,7 @@ namespace System.Reactive
         public IDisposable SubscribeRaw(IObserver<TSource> observer, bool enableSafeguard)
         {
             IDisposable run;
-            ISafeObserver<TSource> safeObserver = null;
+            ISafeObserver<TSource>? safeObserver = null;
 
             //
             // See AutoDetachObserver.cs for more information on the safeguarding requirement and
@@ -101,7 +99,7 @@ namespace System.Reactive
 
         public IDisposable SubscribeRaw(IObserver<TTarget> observer, bool enableSafeguard)
         {
-            ISafeObserver<TTarget> safeObserver = null;
+            ISafeObserver<TTarget>? safeObserver = null;
 
             //
             // See AutoDetachObserver.cs for more information on the safeguarding requirement and

Rx.NET/Source/src/System.Reactive/Internal/PushPullAdapter.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
-
 #if NO_PERF
 using System.Collections.Generic;
 
@@ -80,4 +78,4 @@ namespace System.Reactive
         }
     }
 }
-#endif
+#endif

Rx.NET/Source/src/System.Reactive/Internal/QueryServices.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.PlatformServices;
 
 namespace System.Reactive.Linq

Rx.NET/Source/src/System.Reactive/Internal/SafeObserver.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
@@ -68,7 +66,7 @@ namespace System.Reactive
             return new WrappingSafeObserver(observer);
         }
 
-        private IDisposable _disposable;
+        private IDisposable? _disposable;
 
         public abstract void OnNext(TSource value);
 

Rx.NET/Source/src/System.Reactive/Internal/ScheduledObserver.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;
 using System.Threading;
@@ -22,12 +20,12 @@ namespace System.Reactive
         private const int Faulted = 9;
         private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
         private bool _failed;
-        private Exception _error;
+        private Exception? _error;
         private bool _completed;
         private readonly IObserver<T> _observer;
         private readonly IScheduler _scheduler;
-        private readonly ISchedulerLongRunning _longRunning;
-        private IDisposable _disposable;
+        private readonly ISchedulerLongRunning? _longRunning;
+        private IDisposable? _disposable;
 
         public ScheduledObserver(IScheduler scheduler, IObserver<T> observer)
         {
@@ -44,11 +42,11 @@ namespace System.Reactive
 
         private sealed class SemaphoreSlimRelease : IDisposable
         {
-            private SemaphoreSlim _dispatcherEvent;
+            private volatile SemaphoreSlim? _dispatcherEvent;
 
             public SemaphoreSlimRelease(SemaphoreSlim dispatcherEvent)
             {
-                Volatile.Write(ref _dispatcherEvent, dispatcherEvent);
+                _dispatcherEvent = dispatcherEvent;
             }
 
             public void Dispose()
@@ -58,9 +56,9 @@ namespace System.Reactive
         }
 
         private readonly object _dispatcherInitGate = new object();
-        private readonly SemaphoreSlim _dispatcherEvent;
-        private readonly IDisposable _dispatcherEventRelease;
-        private IDisposable _dispatcherJob;
+        private readonly SemaphoreSlim? _dispatcherEvent;
+        private readonly IDisposable? _dispatcherEventRelease;
+        private IDisposable? _dispatcherJob;
 
         private void EnsureDispatcher()
         {
@@ -86,7 +84,7 @@ namespace System.Reactive
         {
             while (true)
             {
-                _dispatcherEvent.Wait();
+                _dispatcherEvent!.Wait(); // NB: If long-running, the event is set.
 
                 if (cancel.IsDisposed)
                 {
@@ -118,7 +116,7 @@ namespace System.Reactive
 
                 if (_failed)
                 {
-                    _observer.OnError(_error);
+                    _observer.OnError(_error!);
                     Dispose();
                     return;
                 }
@@ -140,7 +138,7 @@ namespace System.Reactive
             {
                 if (n > 0)
                 {
-                    _dispatcherEvent.Release(n);
+                    _dispatcherEvent!.Release(n); // NB: If long-running, the event is set.
                 }
 
                 EnsureDispatcher();
@@ -200,11 +198,11 @@ namespace System.Reactive
 
             if (isOwner)
             {
-                Disposable.TrySetSerial(ref _disposable, _scheduler.Schedule<object>(null, Run));
+                Disposable.TrySetSerial(ref _disposable, _scheduler.Schedule<object?>(null, Run));
             }
         }
 
-        private void Run(object state, Action<object> recurse)
+        private void Run(object? state, Action<object?> recurse)
         {
             T next;
 
@@ -232,7 +230,7 @@ namespace System.Reactive
                     }
 
                     Interlocked.Exchange(ref _state, Stopped);
-                    _observer.OnError(_error);
+                    _observer.OnError(_error!);
                     Dispose();
                     return;
                 }
@@ -326,7 +324,7 @@ namespace System.Reactive
 
     internal sealed class ObserveOnObserver<T> : ScheduledObserver<T>
     {
-        private IDisposable _run;
+        private IDisposable? _run;
 
         public ObserveOnObserver(IScheduler scheduler, IObserver<T> observer)
             : base(scheduler, observer)
@@ -388,7 +386,7 @@ namespace System.Reactive
         /// <summary>
         /// The current task representing a running drain operation.
         /// </summary>
-        private IDisposable _task;
+        private IDisposable? _task;
 
         /// <summary>
         /// Indicates the work-in-progress state of this operator,
@@ -405,7 +403,7 @@ namespace System.Reactive
         /// If <see cref="_done"/> is true and this is non-null, the upstream
         /// failed with an OnError.
         /// </summary>
-        private Exception _error;
+        private Exception? _error;
 
         /// <summary>
         /// Indicates a dispose has been requested.
@@ -558,25 +556,20 @@ namespace System.Reactive
             }
 
             // get the next item from the queue if any
-            var empty = !q.TryDequeue(out var v);
+            if (q.TryDequeue(out var v))
+            {
+                ForwardOnNext(v);
+                return;
+            }
 
             // the upstream called OnComplete and the queue is empty
             // that means we are done, no further signals can happen
-            if (d && empty)
+            if (d)
             {
                 Volatile.Write(ref _disposed, true);
                 // otherwise, complete normally
                 ForwardOnCompleted();
-                return;
-            }
-            
-            // the queue is empty and the upstream hasn't completed yet
-            if (empty)
-            {
-                return;
             }
-            // emit the item
-            ForwardOnNext(v);
         }
     }
 
@@ -617,7 +610,7 @@ namespace System.Reactive
         /// <summary>
         /// Set to a non-null Exception if the upstream terminated with OnError.
         /// </summary>
-        private Exception _error;
+        private Exception? _error;
 
         /// <summary>
         /// Indicates the sequence has been disposed and the drain task should quit.
@@ -633,7 +626,7 @@ namespace System.Reactive
         /// <summary>
         /// The disposable tracking the drain task.
         /// </summary>
-        private IDisposable _drainTask;
+        private IDisposable? _drainTask;
 
         public ObserveOnObserverLongRunning(ISchedulerLongRunning scheduler, IObserver<TSource> observer) : base(observer)
         {
@@ -736,7 +729,7 @@ namespace System.Reactive
                 // There was an item, signal it.
                 if (hasValue)
                 {
-                    ForwardOnNext(item);
+                    ForwardOnNext(item!);
                     // Consume the item and try the next item if the work-in-progress
                     // indicator is still not zero
                     if (Interlocked.Decrement(ref _wip) != 0L)

Rx.NET/Source/src/System.Reactive/Internal/Sink.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;
 using System.Threading;
 
@@ -18,7 +16,7 @@ namespace System.Reactive
 
     internal abstract class Sink<TTarget> : ISink<TTarget>, IDisposable
     {
-        private IDisposable _upstream;
+        private IDisposable? _upstream;
         private volatile IObserver<TTarget> _observer;
 
         protected Sink(IObserver<TTarget> observer)

Rx.NET/Source/src/System.Reactive/Internal/SynchronizationContextExtensions.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.Threading;
 
 namespace System.Reactive.Concurrency
@@ -19,7 +17,7 @@ namespace System.Reactive.Concurrency
                 {
                     try
                     {
-                        action((T)o);
+                        action((T)o!);
                     }
                     finally
                     {

Rx.NET/Source/src/System.Reactive/Internal/SynchronizedObserver.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
 {
     internal sealed class SynchronizedObserver<T> : ObserverBase<T>

Rx.NET/Source/src/System.Reactive/Internal/SystemClock.Default.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.ComponentModel;
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
@@ -41,7 +39,7 @@ namespace System.Reactive.PlatformServices
     public class PeriodicTimerSystemClockMonitor : INotifySystemClockChanged
     {
         private readonly TimeSpan _period;
-        private IDisposable _timer;
+        private IDisposable? _timer;
 
         /// <summary>
         /// Use the Unix milliseconds for the current time
@@ -49,7 +47,7 @@ namespace System.Reactive.PlatformServices
         /// </summary>
         private long _lastTimeUnixMillis;
 
-        private EventHandler<SystemClockChangedEventArgs> _systemClockChanged;
+        private EventHandler<SystemClockChangedEventArgs>? _systemClockChanged;
 
         private const int SyncMaxRetries = 100;
         private const double SyncMaxDelta = 10;

Rx.NET/Source/src/System.Reactive/Internal/SystemClock.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.ComponentModel;
 using System.Reactive.Concurrency;
@@ -24,7 +22,7 @@ namespace System.Reactive.PlatformServices
         private static readonly Lazy<ISystemClock> ServiceSystemClock = new Lazy<ISystemClock>(InitializeSystemClock);
         private static readonly Lazy<INotifySystemClockChanged> ServiceSystemClockChanged = new Lazy<INotifySystemClockChanged>(InitializeSystemClockChanged);
         internal static readonly HashSet<WeakReference<LocalScheduler>> SystemClockChanged = new HashSet<WeakReference<LocalScheduler>>();
-        private static IDisposable _systemClockChangedHandlerCollector;
+        private static IDisposable? _systemClockChangedHandlerCollector;
 
         private static int _refCount;
 
@@ -57,7 +55,7 @@ namespace System.Reactive.PlatformServices
             }
         }
 
-        internal static void OnSystemClockChanged(object sender, SystemClockChangedEventArgs e)
+        internal static void OnSystemClockChanged(object? sender, SystemClockChangedEventArgs e)
         {
             lock (SystemClockChanged)
             {
@@ -123,7 +121,7 @@ namespace System.Reactive.PlatformServices
             //
             lock (SystemClockChanged)
             {
-                HashSet<WeakReference<LocalScheduler>> remove = null;
+                HashSet<WeakReference<LocalScheduler>>? remove = null;
 
                 foreach (var handler in SystemClockChanged)
                 {
@@ -148,7 +146,7 @@ namespace System.Reactive.PlatformServices
 
                 if (SystemClockChanged.Count == 0)
                 {
-                    _systemClockChangedHandlerCollector.Dispose();
+                    _systemClockChangedHandlerCollector?.Dispose();
                     _systemClockChangedHandlerCollector = null;
                 }
             }

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

@@ -2,9 +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
-
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Reactive.Disposables;
 using System.Threading;
 
@@ -12,16 +11,17 @@ namespace System.Reactive
 {
     internal abstract class TailRecursiveSink<TSource> : IdentitySink<TSource>
     {
+        private readonly Stack<IEnumerator<IObservable<TSource>>> _stack = new Stack<IEnumerator<IObservable<TSource>>>();
+
+        private bool _isDisposed;
+        private int _trampoline;
+        private IDisposable? _currentSubscription;
+
         protected TailRecursiveSink(IObserver<TSource> observer)
             : base(observer)
         {
         }
 
-        private bool _isDisposed;
-        private int _trampoline;
-        private IDisposable _currentSubscription;
-        private readonly Stack<IEnumerator<IObservable<TSource>>> _stack = new Stack<IEnumerator<IObservable<TSource>>>();
-
         public void Run(IEnumerable<IObservable<TSource>> sources)
         {
             if (!TryGetEnumerator(sources, out var current))
@@ -85,7 +85,7 @@ namespace System.Reactive
                             continue;
                         }
 
-                        IObservable<TSource> next;
+                        IObservable<TSource>? next;
 
                         try
                         {
@@ -183,9 +183,9 @@ namespace System.Reactive
             }
         }
 
-        protected abstract IEnumerable<IObservable<TSource>> Extract(IObservable<TSource> source);
+        protected abstract IEnumerable<IObservable<TSource>>? Extract(IObservable<TSource> source);
 
-        private bool TryGetEnumerator(IEnumerable<IObservable<TSource>> sources, out IEnumerator<IObservable<TSource>> result)
+        private bool TryGetEnumerator(IEnumerable<IObservable<TSource>> sources, [NotNullWhen(true)] out IEnumerator<IObservable<TSource>>? result)
         {
             try
             {

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

@@ -1,9 +1,7 @@
-// Licensed to the .NET Foundation under one or more agreements.
+// Licensed to the .NET Foundation under one or more agreements.
 // 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.Threading.Tasks
 {
     internal static class TaskExtensions
@@ -13,7 +11,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },
@@ -25,7 +23,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },
@@ -38,7 +36,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },
@@ -51,7 +49,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },
@@ -66,7 +64,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },
@@ -78,7 +76,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },
@@ -91,7 +89,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },
@@ -104,7 +102,7 @@ namespace System.Threading.Tasks
             return task.ContinueWith(
                 (t, tupleObject) =>
                 {
-                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject;
+                    var (closureAction, closureState) = ((Action<Task<TResult>, TState>, TState))tupleObject!;
 
                     closureAction(t, closureState);
                 },

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

@@ -2359,7 +2359,7 @@ namespace System.Reactive.PlatformServices
     public class CurrentPlatformEnlightenmentProvider : System.Reactive.PlatformServices.IPlatformEnlightenmentProvider
     {
         public CurrentPlatformEnlightenmentProvider() { }
-        public virtual T GetService<T>(object[] args)
+        public virtual T? GetService<T>(object[] args)
             where T :  class { }
     }
     public class DefaultSystemClock : System.Reactive.PlatformServices.ISystemClock
@@ -2373,8 +2373,8 @@ namespace System.Reactive.PlatformServices
     }
     public static class HostLifecycleService
     {
-        public  static  event System.EventHandler<System.Reactive.PlatformServices.HostResumingEventArgs> Resuming;
-        public  static  event System.EventHandler<System.Reactive.PlatformServices.HostSuspendingEventArgs> Suspending;
+        public  static  event System.EventHandler<System.Reactive.PlatformServices.HostResumingEventArgs>? Resuming;
+        public  static  event System.EventHandler<System.Reactive.PlatformServices.HostSuspendingEventArgs>? Suspending;
         public static void AddRef() { }
         public static void Release() { }
     }
@@ -2401,7 +2401,7 @@ namespace System.Reactive.PlatformServices
     }
     public interface IPlatformEnlightenmentProvider
     {
-        T GetService<T>(params object[] args)
+        T? GetService<T>(params object[] args)
             where T :  class;
     }
     public interface ISystemClock