using System;

using System.Threading.Tasks;

using System.Threading.Tasks.Dataflow;

​

public class Program

{

    public class Producer

    {

        private BufferBlock<int> _queue;

        private ActionBlock<int> _consumer;

        public Action<int, string> OnResult;

        public Producer()

        {

            InitializeChain();

        }

​

        private void InitializeChain()

        {

            _queue = new BufferBlock<int>(new DataflowBlockOptions{BoundedCapacity = 5});

            var consumerOptions = new ExecutionDataflowBlockOptions{BoundedCapacity = 5, MaxDegreeOfParallelism = 5};

            _consumer = new ActionBlock<int>(SomeIoWorkAsync, consumerOptions);

            _queue.LinkTo(_consumer, new DataflowLinkOptions{PropagateCompletion = true});

        }

​

        private async Task SomeIoWorkAsync(int x)

        {

            Console.WriteLine($"{DateTime.Now.TimeOfDay:g} : Processing {x}");

            await Task.Delay(200);

            OnResult?.Invoke(x, $"SomeResult {x}");

        }

​

        public async Task AddAsync(int data)

        {

            await _queue.SendAsync(data);

        }

    }

​

    public static Task<string> WaitForConsumerAsync(Producer producer, int myId)

    {

        var tcs = new TaskCompletionSource<string>();

        producer.OnResult += (id, result) =>

        {

            if (id == myId)

                tcs.TrySetResult(result);

        }

​

        ;

        return tcs.Task;

    }

​

    public static async Task Main()

    {

        var producer = new Producer();

        // to simulate something you are waiting for, and id or what ever

        var myId = 7;

        var task = WaitForConsumerAsync(producer, myId);

        // Run this is a task a test so we don't hit the back pressure to await

#pragma warning disable 4014

        Task.Run(async () =>

        {

            // create random work for the bounded capacity to fill up

            for (int i = 1; i <= 20; i++)

            {

                await producer.AddAsync(i);

            }

        });

#pragma warning restore 4014

        // wait for your results to pop out

        var result = await task;

        Console.WriteLine($"{DateTime.Now.TimeOfDay:g} : Got my result {result}, now i can finish happily");

        Console.ReadKey();

    }

} 

​