Как я могу сделать следующее наблюдаемое повторение, пока stream.DataAvailable не станет ложным? В настоящее время похоже, что он никогда не останавливается.
AsyncReadChunk и Observable.Return внутри раздела Defer выполняют вызов OnNext, затем вызов OnCompleted. Когда Repeat получает вызов OnNext, он передает его TakeWhile. Когда TakeWhile не удовлетворен, он завершает наблюдаемое, но я думаю, что OnCompleted, который появляется сразу после OnNext, настолько быстр, что заставляет Repeat повторно подписываться на наблюдаемое и вызывает бесконечный цикл.
Как я могу исправить такое поведение?
public static IObservable<byte[]> AsyncRead(this NetworkStream stream, int bufferSize)
{
return Observable.Defer(() =>
{
try
{
return stream.DataAvailable ? AsyncReadChunk(stream, bufferSize) : Observable.Return(new byte[0]);
}
catch (Exception)
{
return Observable.Return(new byte[0]);
}
})
.Repeat()
.TakeWhile((dataChunk, index) => dataChunk.Length > 0);
}