4

我正在尝试将以下类转换为懒惰地返回文件。

public class ObservableFile2 : IObservable<string>
{
    private readonly IObservable<string> subject;

    public ObservableFile2(string fileName)
    {
        subject = Observable.Using<string, StreamReader>
            (
                () => new StreamReader(new FileStream(fileName, FileMode.Open, FileAccess.Read, FileShare.Read)),
                streamReader => ObserveLines(streamReader)
            );
    }

    private IObservable<string> ObserveLines(StreamReader streamReader)
    {
        return ReadLines(streamReader).ToObservable();
    }

    private IEnumerable<string> ReadLines(StreamReader streamReader)
    {
        while (!streamReader.EndOfStream)
        {
            yield return streamReader.ReadLine();
        }
    }        

    public IDisposable Subscribe(IObserver<string> observer)
    {
        return subject.Subscribe(observer);
    }
}

我现在正在尝试将其转换为使用

StreamReader.ReadLineAsync()

甚至更好的是对数据进行分块，即

await SourceStream.ReadAsync(buffer, 0, chunkSize).

我似乎没有掌握如何包装和打开任务

欢迎提供协助。

谢谢

4

2 回答 2

8

我不是Rx 大师，所以可能有比我的答案更好的方法。

我相信这应该可以使用async-enabled Create：

public static class ObservableFile2
{
  public static IObservable<string> Create(string fileName)
  {
    return Observable.Create<string>(async (subject, token) =>
    {
      try
      {
        using (var streamReader = new StreamReader(new FileStream(fileName, FileMode.Open, FileAccess.Read, FileShare.Read))
        {
          while (true)
          {
            token.ThrowIfCancellationRequested();
            var line = await streamReader.ReadLineAsync();
            if (line == null)
            {
              subject.OnCompleted();
              return;
            }
            subject.OnNext(line);
          }
        }
      }
      catch (Exception ex)
      {
        subject.OnError(ex);
      }
    });
  }
}
于 2012-10-31T15:44:32.680 回答
0

我不知道您的解决方案是否需要异步功能，因为您没有提到它-我只能看到您想“懒惰地”使用该文件-我的猜测是您想获得第一行一个时间，如果是这样，这个应该可以解决问题：

public static IEnumerable<string> EnumerateLines(string fileName)
{
    using (
        var streamReader =
            new StreamReader(new FileStream(fileName,
                                            FileMode.Open,
                                            FileAccess.Read,
                                            FileShare.Read)))
    {
        while (true)
        {
            if (streamReader.EndOfStream)
                yield break;

            Console.WriteLine("read another line...");
            yield return streamReader.ReadLine();
        }
    }
}

请注意，这仅取决于 StreamReader.ReadLine 的实现细节。您甚至可以尝试使用 Lazy 对这种非严格评估进行评估 - 但随后您会遇到处理文件句柄的麻烦，因为您不知道这些值何时真正被消耗（即使 Haskell 也有这个问题:)） -我的建议：不要试图对文件懒惰......这会让你很容易陷入困境

于 2012-11-06T09:20:43.227 回答