The ToObservable operator can throw unhandled exception

[theodorzoulias]

Hi, and happy new year! I think that I have found a bug in the ToObservable extension method for async-enumerable sequences. In case an exception is thrown by the GetAsyncEnumerator invocation, the exception is not propagated by the OnError mechanism of the produced observable, and instead it is rethrown as an unhandled exception that crashes the process. Here is a minimal demonstration of this behavior:

using System;
using System.Linq;
using System.Threading;
using System.Collections.Generic;

public class Program
{
    public static void Main()
    {
        var observable = new BadAsyncEnumerable().ToObservable();
        observable.Subscribe(item => { }, error => Console.WriteLine(error.Message));
        Thread.Sleep(1000);
    }

    class BadAsyncEnumerable : IAsyncEnumerable<int>
    {
        public IAsyncEnumerator<int> GetAsyncEnumerator(
            CancellationToken cancellationToken = default)
        {
            throw new Exception("Oops!");
        }
    }
}

Expected behavior: The error message is printed in the console.

Actual behavior: Unhandled exception.

Try it on Fiddle.