SingleOrDefaultAsync.

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

@@ -121,8 +121,8 @@ namespace System.Reactive.Linq
         IObservable<bool> SequenceEqual<TSource>(IObservable<TSource> first, IEnumerable<TSource> second, IEqualityComparer<TSource> comparer);
         IObservable<TSource> SingleAsync<TSource>(IObservable<TSource> source);
         IObservable<TSource> SingleAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
-        IObservable<TSource> SingleOrDefaultAsync<TSource>(IObservable<TSource> source);
-        IObservable<TSource> SingleOrDefaultAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
+        IObservable<TSource?> SingleOrDefaultAsync<TSource>(IObservable<TSource> source);
+        IObservable<TSource?> SingleOrDefaultAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate);
         IObservable<double> Sum(IObservable<double> source);
         IObservable<float> Sum(IObservable<float> source);
         IObservable<decimal> Sum(IObservable<decimal> source);

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

@@ -2332,7 +2332,7 @@ namespace System.Reactive.Linq
         /// <returns>Sequence containing the single element in the observable sequence, or a default value if no such element exists.</returns>
         /// <exception cref="ArgumentNullException"><paramref name="source"/> is null.</exception>
         /// <exception cref="InvalidOperationException">(Asynchronous) The source sequence contains more than one element.</exception>
-        public static IObservable<TSource> SingleOrDefaultAsync<TSource>(this IObservable<TSource> source)
+        public static IObservable<TSource?> SingleOrDefaultAsync<TSource>(this IObservable<TSource> source)
         {
             if (source == null)
             {
@@ -2351,7 +2351,7 @@ namespace System.Reactive.Linq
         /// <returns>Sequence containing the single element in the observable sequence that satisfies the condition in the predicate, or a default value if no such element exists.</returns>
         /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="predicate"/> is null.</exception>
         /// <exception cref="InvalidOperationException">(Asynchronous) The sequence contains more than one element that satisfies the condition in the predicate.</exception>
-        public static IObservable<TSource> SingleOrDefaultAsync<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
+        public static IObservable<TSource?> SingleOrDefaultAsync<TSource>(this IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             if (source == null)
             {

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

@@ -6,7 +6,7 @@ namespace System.Reactive.Linq.ObservableImpl
 {
     internal static class SingleOrDefaultAsync<TSource>
     {
-        internal sealed class Sequence : Producer<TSource, Sequence._>
+        internal sealed class Sequence : Producer<TSource?, Sequence._>
         {
             private readonly IObservable<TSource> _source;
 
@@ -15,16 +15,16 @@ namespace System.Reactive.Linq.ObservableImpl
                 _source = source;
             }
 
-            protected override _ CreateSink(IObserver<TSource> observer) => new _(observer);
+            protected override _ CreateSink(IObserver<TSource?> observer) => new _(observer);
 
             protected override void Run(_ sink) => sink.Run(_source);
 
-            internal sealed class _ : IdentitySink<TSource>
+            internal sealed class _ : Sink<TSource, TSource?>
             {
                 private TSource? _value;
                 private bool _seenValue;
 
-                public _(IObserver<TSource> observer)
+                public _(IObserver<TSource?> observer)
                     : base(observer)
                 {
                 }
@@ -50,13 +50,13 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    ForwardOnNext(_value!);
+                    ForwardOnNext(_value);
                     ForwardOnCompleted();
                 }
             }
         }
 
-        internal sealed class Predicate : Producer<TSource, Predicate._>
+        internal sealed class Predicate : Producer<TSource?, Predicate._>
         {
             private readonly IObservable<TSource> _source;
             private readonly Func<TSource, bool> _predicate;
@@ -67,17 +67,17 @@ namespace System.Reactive.Linq.ObservableImpl
                 _predicate = predicate;
             }
 
-            protected override _ CreateSink(IObserver<TSource> observer) => new _(_predicate, observer);
+            protected override _ CreateSink(IObserver<TSource?> observer) => new _(_predicate, observer);
 
             protected override void Run(_ sink) => sink.Run(_source);
 
-            internal sealed class _ : IdentitySink<TSource>
+            internal sealed class _ : Sink<TSource, TSource?>
             {
                 private readonly Func<TSource, bool> _predicate;
                 private TSource? _value;
                 private bool _seenValue;
 
-                public _(Func<TSource, bool> predicate, IObserver<TSource> observer)
+                public _(Func<TSource, bool> predicate, IObserver<TSource?> observer)
                     : base(observer)
                 {
                     _predicate = predicate;
@@ -119,7 +119,7 @@ namespace System.Reactive.Linq.ObservableImpl
 
                 public override void OnCompleted()
                 {
-                    ForwardOnNext(_value!);
+                    ForwardOnNext(_value);
                     ForwardOnCompleted();
                 }
             }

Rx.NET/Source/src/System.Reactive/Linq/Qbservable.Generated.cs

@@ -12755,7 +12755,7 @@ namespace System.Reactive.Linq
         /// <exception cref="ArgumentNullException">
         /// <paramref name="source" /> is null.</exception>
         /// <exception cref="InvalidOperationException">(Asynchronous) The source sequence contains more than one element.</exception>
-        public static IQbservable<TSource> SingleOrDefaultAsync<TSource>(this IQbservable<TSource> source)
+        public static IQbservable<TSource?> SingleOrDefaultAsync<TSource>(this IQbservable<TSource> source)
         {
             if (source == null)
                 throw new ArgumentNullException(nameof(source));

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

@@ -602,12 +602,12 @@ namespace System.Reactive.Linq
 
         #region + SingleOrDefaultAsync +
 
-        public virtual IObservable<TSource> SingleOrDefaultAsync<TSource>(IObservable<TSource> source)
+        public virtual IObservable<TSource?> SingleOrDefaultAsync<TSource>(IObservable<TSource> source)
         {
             return new SingleOrDefaultAsync<TSource>.Sequence(source);
         }
 
-        public virtual IObservable<TSource> SingleOrDefaultAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
+        public virtual IObservable<TSource?> SingleOrDefaultAsync<TSource>(IObservable<TSource> source, Func<TSource, bool> predicate)
         {
             return new SingleOrDefaultAsync<TSource>.Predicate(source, predicate);
         }

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

@@ -1359,8 +1359,8 @@ namespace System.Reactive.Linq
         [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) { }
+        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) { }
         public static System.IObservable<TSource> Skip<TSource>(this System.IObservable<TSource> source, int count) { }
         public static System.IObservable<TSource> Skip<TSource>(this System.IObservable<TSource> source, System.TimeSpan duration) { }
         public static System.IObservable<TSource> Skip<TSource>(this System.IObservable<TSource> source, System.TimeSpan duration, System.Reactive.Concurrency.IScheduler scheduler) { }
@@ -2183,7 +2183,7 @@ namespace System.Reactive.Linq
         public static System.Reactive.Linq.IQbservable<bool> SequenceEqual<TSource>(this System.Reactive.Linq.IQbservable<TSource> first, System.IObservable<TSource> second, System.Collections.Generic.IEqualityComparer<TSource> comparer) { }
         public static System.Reactive.Linq.IQbservable<TSource> SingleAsync<TSource>(this System.Reactive.Linq.IQbservable<TSource> source) { }
         public static System.Reactive.Linq.IQbservable<TSource> SingleAsync<TSource>(this System.Reactive.Linq.IQbservable<TSource> source, System.Linq.Expressions.Expression<System.Func<TSource, bool>> predicate) { }
-        public static System.Reactive.Linq.IQbservable<TSource> SingleOrDefaultAsync<TSource>(this System.Reactive.Linq.IQbservable<TSource> source) { }
+        public static System.Reactive.Linq.IQbservable<TSource?> SingleOrDefaultAsync<TSource>(this System.Reactive.Linq.IQbservable<TSource> source) { }
         public static System.Reactive.Linq.IQbservable<TSource> SingleOrDefaultAsync<TSource>(this System.Reactive.Linq.IQbservable<TSource> source, System.Linq.Expressions.Expression<System.Func<TSource, bool>> predicate) { }
         public static System.Reactive.Linq.IQbservable<TSource> Skip<TSource>(this System.Reactive.Linq.IQbservable<TSource> source, int count) { }
         public static System.Reactive.Linq.IQbservable<TSource> Skip<TSource>(this System.Reactive.Linq.IQbservable<TSource> source, System.TimeSpan duration) { }