Async Produce Consume Coordinator - ProducerConsumerHub<T>

ProducerConsumerHub_T.cs

using System;
using System.Threading.Tasks;

namespace AsyncReadWritePipeline
{
    public class Program
    {
        public class ProducerConsumerHub<T>
        {
            TaskCompletionSource<Empty> _consumer = new TaskCompletionSource<Empty>();
            TaskCompletionSource<T> _producer = new TaskCompletionSource<T>();

            //[Unsafe]
            public async Task ProduceAsync(T data)
            {
                _producer.SetResult(data);
                await _consumer.Task;
                _consumer = new TaskCompletionSource<Empty>();
            }

            public async Task<T> ConsumeAsync()
            {
                var data = await _producer.Task;
                _producer = new TaskCompletionSource<T>();
                _consumer.SetResult(Empty.Value);
                return data;
            }

            struct Empty { 
               public static readonly Empty Value = default(Empty);
            }
        }    
    
        public static void Main(string[] args)
        {
            new Program().ConsumeSomeDataAsync().Wait();
        }

        public async Task GetSomeDataAsync(ProducerConsumerHub<int> hub)
        {
            for (int i = 0; i < 10; i++)
            {
                await hub.ProduceAsync(i);
            }
        }

        public async Task ConsumeSomeDataAsync()
        {
            var hub = new ProducerConsumerHub<int>();
            var producerTask = GetSomeDataAsync(hub);

            while (true)
            {
                var dataItemTask = hub.ConsumeAsync();
                await Task.WhenAny(producerTask, dataItemTask);

                if (dataItemTask.IsCompleted)
                {
                    // process data item
                    Console.WriteLine(await dataItemTask);
                }
                if (producerTask.IsCompleted)
                {
                    // process the end
                    await producerTask;
                    break;
                }
            }
        }
    }
}