title: Combination of Observable.FromAsync+Repeat+TakeWhile creates infinite loop
body: <p>Can someone explain why the following <code>AsObservable</code> method creates an infinite loop even though the end of stream is reached?</p>
  
  <pre class="lang-cs prettyprint-override"><code>public static class StreamExt {
      public static IObservable&lt;byte&gt; AsObservable(this Stream stream, int bufferSize) {
          return Observable
              .FromAsync(cancel =&gt; stream.ReadBytes(bufferSize, cancel))
              .Repeat()
              .TakeWhile(bytes =&gt; bytes != null) // EndOfStream
              .SelectMany(bytes =&gt; bytes);
      }
  
      private static async Task&lt;byte[]&gt; ReadBytes(this Stream stream, int bufferSize, CancellationToken cancel) {
          var buf = new byte[bufferSize];
          var bytesRead = await stream
              .ReadAsync(buf, 0, bufferSize, cancel)
              .ConfigureAwait(false);
  
          if (bytesRead &lt; 1) return null; // EndOfStream
          var result_size = Math.Min(bytesRead, bufferSize);
          Array.Resize(ref buf, result_size);
          return buf;
      }
  }
  </code></pre>
  
  <p>A quick tests shows that it produces an infinite loop:</p>
  
  <pre class="lang-cs prettyprint-override"><code>class Program {
      static void Main(string[] args) {
          using (var stream = new MemoryStream(new byte[] { 1, 2, 3 })) {
              var testResult = stream
                  .AsObservable(1024)
                  .ToEnumerable()
                  .ToArray();
              Console.WriteLine(testResult.Length);
          }
      }
  }
  </code></pre>
  
  <p>Of course I could add an <code>.SubscribeOn(TaskPoolScheduler.Default)</code> but however, the infinite loop stays alive (blocks a task pool scheduler + infinitely reads from <code>Stream</code>).</p>
  
  <p><strong>[UPDATE 2017-05-09]</strong></p>
  
  <p>Shlomo posted a better example to reproduce this issue:</p>
  
  <pre class="lang-cs prettyprint-override"><code>int i = 0;
  var testResult = Observable.FromAsync(() =&gt; Task.FromResult(i++))
      .Repeat()
      .TakeWhile(l =&gt; l &lt; 3);
  testResult.Subscribe(b =&gt; Console.WriteLine(b), e =&gt; { }, () =&gt; Console.WriteLine("OnCompleted"));
  Console.WriteLine("This is never printed.");
  </code></pre>