title: System.Threading.Channels ReadAsync() method is blocking execution
body: <p><strong>Overview</strong></p>
  <p>I am attempting to write an <code>IAsyncEnumerable&lt;T&gt;</code> wrapper around an <code>IObserver&lt;T&gt;</code> interface. At first I used a <code>BufferBlock&lt;T&gt;</code> as the backing data store, but I found out through performance testing and research that it is actually a pretty slow type, so I decided to give the <code>System.Threading.Channels.Channel</code> type a go. I had a similar problem with my BufferBlock implementation as this one but this time I'm not sure how to resolve it.</p>
  <p><strong>Problem</strong></p>
  <p>My <code>GetAsyncEnumerator()</code> loop gets blocked by the <code>await _channel.Reader.WaitToRead(token)</code> call if my <code>IObserver&lt;T&gt;.OnNext()</code> method hasn't written to the <code>_channel</code> yet. What is the correct way to wait for a value to be available to yield in this context without blocking program execution?</p>
  <p><strong>Implementation</strong></p>
  <pre class="lang-cs prettyprint-override"><code>public sealed class ObserverAsyncEnumerableWrapper&lt;T&gt; : IAsyncEnumerable&lt;T&gt;,
      IObserver&lt;T&gt;, IDisposable
  {
      private readonly IDisposable _unsubscriber;
      private readonly Channel&lt;T&gt; _channel = Channel.CreateUnbounded&lt;T&gt;();
  
      private bool _producerComplete;
  
      public ObserverAsyncEnumerableWrapper(IObservable&lt;T&gt; provider)
      {
          _unsubscriber = provider.Subscribe(this);
      }
  
      public async void OnNext(T value)
      {
          Log.Logger.Verbose(&quot;Adding value to Channel.&quot;);
          await _channel.Writer.WriteAsync(value);
      }
  
      public void OnError(Exception error)
      {
          _channel.Writer.Complete(error);
      }
  
      public void OnCompleted()
      {
          _producerComplete = true;
      }
  
      public async IAsyncEnumerator&lt;T&gt; GetAsyncEnumerator(
          [EnumeratorCancellation] CancellationToken token = new CancellationToken())
      {
          Log.Logger.Verbose(&quot;Starting async iteration...&quot;);
          while (await _channel.Reader.WaitToReadAsync(token) || !_producerComplete)
          {
              Log.Logger.Verbose(&quot;Reading...&quot;);
              while (_channel.Reader.TryRead(out var item))
              {
                  Log.Logger.Verbose(&quot;Yielding item.&quot;);
                  yield return item;
              }
              Log.Logger.Verbose(&quot;Awaiting more items.&quot;);
          }
          Log.Logger.Verbose(&quot;Iteration Complete.&quot;);
          _channel.Writer.Complete();
      }
  
      public void Dispose()
      {
          _channel.Writer.Complete();
          _unsubscriber?.Dispose();
      }
  }
  </code></pre>
  <p><a href="https://i.stack.imgur.com/5tnJQ.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/5tnJQ.png" alt="Screenshot" /></a></p>
  <p><strong>Additional Context</strong></p>
  <p>It shouldn't matter, but at runtime the <code>IObservable&lt;T&gt;</code> instance passed into the constructor is a <code>CimAsyncResult</code> returned from async calls made to the <code>Microsoft.Management.Infrastructure</code> apis. Those make use of the Observer design pattern which I'm trying to wrap with the fancy new async enumeration pattern.</p>
  <p><strong>Edit</strong></p>
  <p>Updated with logging to the debugger output and made my <code>OnNext()</code> method async/await as one commenter suggested. You can see it never enters the <code>while()</code> loop.</p>