可讀的流，直到它結束或出現錯誤

Question

如何使用反應式擴展，如何創建一個Observable，它將持續調用流的Read方法並將結果傳播給其觀察者？

或者這是接近事物的完全錯誤的方式？我應該實施自己的IObservable嗎？

Answer 1

我從來沒有遇到過實施我自己的observable有意義的情況。

試試這個：

public static IObservable<byte[]> ObservableRead(Stream stream, int bufferSize) 
{ 
    return Observable.Create<byte[]>(o => 
    { 
     var buffer = new byte[bufferSize]; 
     var read = 0; 
     try 
     { 
      while (true) 
      { 
       read = stream.Read(buffer, 0, buffer.Length); 
       if (read == 0) 
       { 
        break; 
       } 
       var results = buffer.Take(read).ToArray(); 
       //Always return a copy 
       //never the buffer for concurrency's sake. 
       o.OnNext(results); 
      } 
     } 
     catch (Exception ex) 
     { 
      o.OnError(ex); 
     } 
     finally 
     { 
      o.OnCompleted(); 
     } 
     return Disposable.Empty; 
    }); 
}