A channel is a data structure that’s used to store produced data for a consumer
to retrieve, and an appropriate synchronization to enable that to happen safely,
while also enabling appropriate notifications in both directions.
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;
}
}