using System;

using System;

using System.Threading;

using System.Threading.Channels;

using System.Threading.Tasks;

​

 public class DummyData

    {

        public long Ticks { get; set; }

        public DateTime DateTime { get; set; }

        public decimal Price { get; set; }

    }

​

​

    class ConsumerOne

    {

        private readonly Channel<DummyData> _tickQueue;

        private readonly CancellationTokenSource _cancellationTokenSource;

        private readonly string _tag;

​

        public ConsumerOne(Channel<DummyData> tickQueue, CancellationTokenSource cancellationTokenSource, string tag)

        {

            _tickQueue = tickQueue;

            _cancellationTokenSource = cancellationTokenSource;

            _tag = tag;

        }

​

        public async Task StartConsuming()

        {

            await foreach (var message in _tickQueue.Reader.ReadAllAsync(

                               cancellationToken: _cancellationTokenSource.Token))

            {

                // Business logic of One

                Console.WriteLine($"from consumer {_tag} ==> {message.Price}");

            }

        }

    }

    class Producer

    {

        private readonly Channel<DummyData> _tickQueue;

        private readonly CancellationTokenSource _cancellationTokenSource;

​

        public Producer(Channel<DummyData> tickQueue, CancellationTokenSource cancellationTokenSource)

        {

            _tickQueue = tickQueue;

            _cancellationTokenSource = cancellationTokenSource;

        }

​

        public async Task StartProducing()

        {

            Random r = new Random();

            for (int i = 0; i < 10; i++)

            {

​

                await _tickQueue.Writer.WriteAsync(new DummyData()

                {

                    DateTime = DateTime.Now,

                    Ticks = DateTime.Now.Ticks,

                    Price = Convert.ToDecimal(r.NextDouble() * r.Next(100, 105))

                }, _cancellationTokenSource.Token);

                await Task.Delay(r.Next(50, 500));

            }

        }

    }

​

    internal class MultipleConsumersEg

    {

        private static Channel<DummyData> tickQueue;

        private static readonly CancellationTokenSource TickQueueCancellationTokenSource = new CancellationTokenSource();

        public static async Task Main(string[] args)

        {

            tickQueue = Channel.CreateUnbounded<DummyData>();

​

            Producer p = new Producer(tickQueue, TickQueueCancellationTokenSource);

            ConsumerOne consumerOne = new ConsumerOne(tickQueue, TickQueueCancellationTokenSource, "ONE");

             consumerOne.StartConsuming();

​

             p.StartProducing();

​

            Console.ReadLine();

        }

    }

​