将Stream转换为Observable时遭遇并发问题的技术求助

大家好，我写了一段把Stream转换成IObservable<byte>的扩展方法，但运行起来好像有并发问题，我现在完全搞不清楚到底哪里出问题了，有没有大佬能帮忙看看？

我的代码是这样的：

public static IObservable<byte> ToObservable(this Stream stream, int bufferSize = 4096)
{
    var buffer = new byte[bufferSize];
    return Observable
        .FromAsync(async ct => (bytesRead: await stream.ReadAsync(buffer, 0, buffer.Length, ct).ConfigureAwait(false), buffer))
        .Repeat()
        .TakeWhile(x => x.bytesRead != 0)
        .Select(x => x.buffer)
        .SelectMany(x => x);
}

麻烦各位帮我分析下问题所在，谢谢啦！

备注：内容来源于stack exchange，提问作者SuperJMN