AsyncObseravableBase: normalized synchronous exceptions thrown by observer methods

Author: fedeAlterio

AsyncRx.NET/System.Reactive.Async/AsyncObservableBase.cs

@@ -2,6 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information. 
 
+using System.Reactive.Internal;
 using System.Threading.Tasks;
 
 namespace System.Reactive
@@ -124,7 +125,7 @@ protected override async ValueTask OnCompletedAsyncCore()
                         return;
                     }
 
-                    _task = _observer.OnCompletedAsync();
+                    _task = _observer.OnCompletedAsync_EnsureAsync();
                 }
 
                 try
@@ -146,7 +147,7 @@ protected override async ValueTask OnErrorAsyncCore(Exception error)
                         return;
                     }
 
-                    _task = _observer.OnErrorAsync(error);
+                    _task = _observer.OnErrorAsync_EnsureAsync(error);
                 }
 
                 try
@@ -168,7 +169,7 @@ protected override async ValueTask OnNextAsyncCore(T value)
                         return;
                     }
 
-                    _task = _observer.OnNextAsync(value);
+                    _task = _observer.OnNextAsync_EnsureAsync(value);
                 }
 
                 try

AsyncRx.NET/System.Reactive.Async/Internal/AsyncObserverEnsureAsyncHelpers.cs

@@ -0,0 +1,44 @@
+using System.Threading.Tasks;
+
+namespace System.Reactive.Internal;
+
+// Helpers methods that ensure that calls to IAsyncObserver methods don't throw synchronously.
+// Those methods will always return a ValueTask, and any exception will be propagated through that ValueTask.
+internal static class AsyncObserverEnsureAsyncHelpers
+{
+    public static ValueTask OnNextAsync_EnsureAsync<T>(this IAsyncObserver<T> source, T value)
+    {
+        try
+        {
+            return source.OnNextAsync(value);
+        }
+        catch (Exception e)
+        {
+            return new ValueTask(Task.FromException(e));
+        }
+    }
+
+    public static ValueTask OnErrorAsync_EnsureAsync<T>(this IAsyncObserver<T> source, Exception error)
+    {
+        try
+        {
+            return source.OnErrorAsync(error);
+        }
+        catch (Exception e)
+        {
+            return new ValueTask(Task.FromException(e));
+        }
+    }
+
+    public static ValueTask OnCompletedAsync_EnsureAsync<T>(this IAsyncObserver<T> source)
+    {
+        try
+        {
+            return source.OnCompletedAsync();
+        }
+        catch (Exception e)
+        {
+            return new ValueTask(Task.FromException(e));
+        }
+    }
+}