using System;

using System.Collections.Generic;

using System.Threading;

using System.Threading.Channels;

using System.Threading.Tasks;

​

public class Program

{

    public static void Main()

    {

        List<Channel<int>> channels = new();

​

        async Task CreateConsumer(Func<Channel<int>, Task> body)

        {

            var channel = Channel.CreateUnbounded<int>();

            lock (channels) channels.Add(channel);

            try

            {

                await Task.Run(() => body(channel)).ConfigureAwait(false);

            }

            finally

            {

                lock (channels) channels.Remove(channel);

            }

        }

​

        Task consumer1 = CreateConsumer(async channel =>

        {

            await foreach (var i in channel.Reader.ReadAllAsync())

            {

                Console.WriteLine($"Consumer one: {i}");

            }

        });

​

        Task consumer2 = CreateConsumer(async channel =>

        {

            await foreach (var i in channel.Reader.ReadAllAsync())

            {

                Console.WriteLine($"Consumer two: {i}");

            }

        });

​

        using CancellationTokenSource cts = new(TimeSpan.FromMilliseconds(3000));

        Task producer = Task.Run(async () =>

        {

            int i = 0;

            while (true)

            {

                i++;

                lock (channels) channels.ForEach(channel => channel.Writer.TryWrite(i));

                try { await Task.Delay(TimeSpan.FromMilliseconds(250), cts.Token); }

                catch (OperationCanceledException) { break; }

            }

        });

​

        producer.Wait();

        lock (channels) channels.ForEach(channel => channel.Writer.Complete());

        Task.WaitAll(consumer1, consumer2);

    }

}

​