c# - 从 Stream 生成 IObservable 的首选方法

Question

作为我们应用程序的一部分(目前已投入生产约 4 个月)，我们有来自外部设备的数据流，我们将其转换为 IObservable

到目前为止，我们一直在使用以下内容生成它，并且效果很好。

IObservable<string> ObserveStringStream(Stream inputStream)
{
    var streamReader = new StreamReader(inputStream);
    return Observable
            .Create<string>(observer => Scheduler.ThreadPool
            .Schedule(() => ReadLoop(streamReader, observer)));
}

private void ReadLoop(StreamReader reader, IObserver<string> observer)
{
    while (true)
    {
        try
        {
            var line = reader.ReadLine();
            if (line != null)
            {
                observer.OnNext(line);
            }
            else
            {
                observer.OnCompleted();
                break;
            }
        }
        catch (Exception ex)
        {
            observer.OnError(ex);
            break;
        }
    }
}

昨晚我想知道是否有一种方法可以使用 yield return 语法来实现相同的结果并想出了这个:

IObservable<string> ObserveStringStream(Stream inputStream)
{
    var streamReader = new StreamReader(inputStream);
    return ReadLoop(streamReader)
            .ToObservable(Scheduler.ThreadPool);
}

private IEnumerable<string> ReadLoop(StreamReader reader)
{
    while (true)
    {
        var line = reader.ReadLine();
        if (line != null)
        {
            yield return line;
        }
        else
        {
            yield break;
        }
    }
}

它似乎工作得很好而且更干净，但我想知道一种方式是否比另一种方式有任何优点或缺点，或者是否有更好的方法。

Answer 1

我认为你有一个好主意(将 Stream 变成 Enumerable<string> 然后 IObservable<string> )。但是，IEnumerable 代码可以更简洁:

IEnumerable<string> ReadLines(Stream stream)
{
    using (StreamReader reader = new StreamReader(stream))
    {
        while (!reader.EndOfStream)
            yield return reader.ReadLine();
    }
}

然后对于 IObservable:

IObservable<string> ObserveLines(Stream inputStream)
{
    return ReadLines(inputStream).ToObservable(Scheduler.ThreadPool);
}

这更短、更易读，并且正确处理了流。它也很懒惰。

ToObservable扩展负责捕捉 OnNext事件(新行)以及 OnCompleted事件(可枚举结束)和 OnError .