Channels.linq

<Query Kind="Program">
  <Namespace>System.Threading.Channels</Namespace>
  <Namespace>System.Threading.Tasks</Namespace>
</Query>

async Task Main()
{
	//Scenario 1: Single producer, single consumer
	await SingleProducerSingleConsumer();
	
	//Scenario 2: Multiple producers, single consumer
	await MultiProducerSingleConsumer();
	
	//Scenario 3: Single producer, multiple consumers
	await SingleProduceMultipleConsumers();
}
public async Task SingleProducerSingleConsumer()
{
	var channel = Channel.CreateUnbounded<string>();

	// In this example, the consumer keeps up with the producer

	var producer1 = new Producer(channel.Writer, 1, 2000);
	var consumer1 = new Consumer(channel.Reader, 1, 1500);

	Task consumerTask1 = consumer1.ConsumeData(); // begin consuming
	Task producerTask1 = producer1.BeginProducing(); // begin producing

	await producerTask1.ContinueWith(_ => channel.Writer.Complete());

	await consumerTask1;
}

public async Task MultiProducerSingleConsumer()
{
	var channel = Channel.CreateUnbounded<string>();

	// In this example, a single consumer easily keeps up with two producers

	var producer1 = new Producer(channel.Writer, 1, 2000);
	var producer2 = new Producer(channel.Writer, 2, 2000);
	var consumer1 = new Consumer(channel.Reader, 1, 250);

	Task consumerTask1 = consumer1.ConsumeData(); // begin consuming

	Task producerTask1 = producer1.BeginProducing();

	await Task.Delay(500); // stagger the producers

	Task producerTask2 = producer2.BeginProducing();

	await Task.WhenAll(producerTask1, producerTask2)
		.ContinueWith(_ => channel.Writer.Complete());

	await consumerTask1;
}

public async Task SingleProduceMultipleConsumers()
{
	var channel = Channel.CreateUnbounded<string>();

	// In this example, multiple consumers are needed to keep up with a fast producer

	var producer1 = new Producer(channel.Writer, 1, 100);
	var consumer1 = new Consumer(channel.Reader, 1, 1500);
	var consumer2 = new Consumer(channel.Reader, 2, 1500);
	var consumer3 = new Consumer(channel.Reader, 3, 1500);

	Task consumerTask1 = consumer1.ConsumeData(); // begin consuming
	Task consumerTask2 = consumer2.ConsumeData(); // begin consuming
	Task consumerTask3 = consumer3.ConsumeData(); // begin consuming

	Task producerTask1 = producer1.BeginProducing();

	await producerTask1.ContinueWith(_ => channel.Writer.Complete());

	await Task.WhenAll(consumerTask1, consumerTask2, consumerTask3);
}

internal class Producer
{
	private readonly ChannelWriter<string> _writer;
	private readonly int _identifier;
	private readonly int _delay;

	public Producer(ChannelWriter<string> writer, int identifier, int delay)
	{
		_writer = writer;
		_identifier = identifier;
		_delay = delay;
	}

	public async Task BeginProducing()
	{
		($"PRODUCER ({_identifier}): Starting").Dump("Producer");

		for (var i = 0; i < 10; i++)
		{
			await Task.Delay(_delay); // simulate producer building/fetching some data

			var msg = $"P{_identifier} - {DateTime.UtcNow:G}";

			($"PRODUCER ({_identifier}): Creating {msg}").Dump("Producer");

			await _writer.WriteAsync(msg);
		}

		($"PRODUCER ({_identifier}): Completed").Dump("Producer");
	}
}


internal class Consumer
{
	private readonly ChannelReader<string> _reader;
	private readonly int _identifier;
	private readonly int _delay;

	public Consumer(ChannelReader<string> reader, int identifier, int delay)
	{
		_reader = reader;
		_identifier = identifier;
		_delay = delay;
	}

	public async Task ConsumeData()
	{
		($"CONSUMER ({_identifier}): Starting").Dump("Consumer");

		while (await _reader.WaitToReadAsync())
		{
			if (_reader.TryRead(out var timeString))
			{
				await Task.Delay(_delay); // simulate processing time

				($"CONSUMER ({_identifier}): Consuming {timeString}").Dump("Consumer");;
			}
		}

		($"CONSUMER ({_identifier}): Completed").Dump("Consumer");;
	}
}