ChannelReader as IObservable

ChannelReaderExtensions.cs

using System.Reactive.Linq;
using System.Threading.Channels;

public static class ChannelReaderExtensions
{
    public static IObservable<T> ToObservable<T>(this ChannelReader<T> reader)
    {
        return Observable.Create<T>(async (observer, ct) =>
        {
            try
            {
                for (;;)
                {
                    var item = await reader.ReadAsync(ct);
                    observer.OnNext(item);
                }
            }
            catch (ChannelClosedException)
            {
                observer.OnCompleted();
            }
            catch (OperationCanceledException)
            {
                observer.OnCompleted();
            }
            catch (Exception exception)
            {
                observer.OnError(exception);
            }
        });
    }
}

ChannelReaderObservable.md

Intention

Wraps a ChannelReader<T> in an IObservable<T>

License

MIT

Usage

using System.Reactive.Linq;
using System.Threading.Channels;

var channel = Channel.CreateBounded<int>(1);

var reader = channel.Reader.ToObservable();

using (reader
           .Do(x => Console.WriteLine($"Received {x}"))
           .Subscribe())
{
    for (var i = 0; i != 1000; ++i)
    {
        Console.WriteLine($"Writing {i}");
        await channel.Writer.WriteAsync(i);
        Console.WriteLine($"Written {i}");
    }

    channel.Writer.Complete();

    Console.WriteLine("Press key to end");
    Console.ReadLine();
}

Console.WriteLine("Finished");