我正试图将下面的类转换为懒洋洋地返回一个文件。
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).
我好像还不知道该怎么做
包装和展开任务
欢迎提供援助。
谢谢
最佳答案
我不是RX大师,所以可能有比我的答案更好的方法。
我相信这应该是可能的,使用async
-enabledCreate
:
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);
}
});
}
}