ストリームから非同期的に繰り返し読み込みますか？

Question

次のコードを構造化して、すべてのストリームをチャンクで読み取ることができます。 現在、同じデータを返し続けます。進まない。

public static IObservable<byte[]> AsyncRead(this Stream stream, int bufferSize) 
{ 
    var asyncRead = Observable.FromAsyncPattern<byte[], int, int, int>(stream.BeginRead, stream.EndRead); 

    var buffer = new byte[bufferSize]; 

    return asyncRead(buffer, 0, bufferSize) 
     .Select(cbRead => 
        { 
         var dataChunk = new byte[cbRead]; 

         Buffer.BlockCopy(buffer, 0, dataChunk, 0, cbRead); 

         return dataChunk; 
        }) 
     .Repeat() 
     .TakeWhile(dataChunk => dataChunk.Length > 0); 
} 

Answer 1

同様のトラップにも入り込みます。非同期ストリームは、再生可能なストリームのように動作します。それはその1つの項目だけを含んでいますが。これを修正するには、Deferのようにこれを囲みます。

Observable.Defer(() => asyncRead(buffer, 0, bufferSize)) 
...