C#5 ReadAsync and Iterators

async-await, c#, c#-5.0, iterator, system.reactive

Solution

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

Problem

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 Assistance would be welcomed. Thanks

Original source