C＃5 ReadAsync和迭代器 [英] C#5 ReadAsync and Iterators

问题描述

我想下面的类转换懒洋洋地返回一个文件。

I am trying to convert the below class to lazily return a file.

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);
    }
}

我现在想将它转换为用

I am now trying to convert this to use

StreamReader.ReadLineAsync() 

甚至更好的将大块的数据，即

or even better would be to chunk the data i.e.

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

我似乎没有掌握如何
包裹和解开任务

I seem to not have grasped how to wrap and unwrap Tasks

援助将受到欢迎。

感谢

推荐答案

我的不的一个接收高手，所以有可能会比我的答案更好的办法。

I am not an Rx master, so there may be a better way than my answer.

我相信这应该使用有可能在异步 - 启用创建：

I believe this should be possible using the 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);
      }
    });
  }
}

这篇关于C＃5 ReadAsync和迭代器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！