Stream Processing

public sealed class Channel<T> {
  // Use a thread-safe data store to free us from locking semantics for any
  // number of producers and consumers.
  private readonly ConcurrentQueue<T> _items = new ConcurrentQueue<T>();

  // A mechanism for coordinating between producers and consumers.
  private readonly SemaphoreSlim _semaphore = new Semaphore(0);

  public void Write(T item)
  {
    _items.Enqueue(item); // Store the data.
    _semaphore.Release(); // Notify any consumers that more data is available.
  }

  public ValueTask<T> ReadAsync(CancellationToken ct = default)
  {
    await _semaphore.WaitAsync(ct).ConfigureAwait(false); // Wait for data to be available.
    bool gotOne = _items.TryDequeue(out T item);
    Debug.Assert(gotOne);
    return item;
  }
}