using System;

using System.Threading.Tasks;

using System.Threading.Tasks.Dataflow;

using System.Collections.Generic;

using System.Linq;

public class Program

{

    public static async Task Main()

    {

        // Construct a new pipeline for each request

        var (input, completion, output) = CreatePipeline();

        IEnumerable<int> data = Enumerable.Repeat(1, 50);

        foreach (int d in data)

        {

            await input.SendAsync(d);

        }

        input.Complete();

        await completion;

        output.Dump();

    }

    static (ITargetBlock<int>, Task, List<string>) CreatePipeline()

    {

        // Set up block to call the API

        var callBlock = new TransformBlock<int, string>(

            data => CallApi(data),

            new ExecutionDataflowBlockOptions()

            {

                MaxDegreeOfParallelism = 5 // Set how many calls will run in parallel

            });

        // Use a list to collect the output data

        var outputData = new List<string>();

        var aggregateBlock = new ActionBlock<string>(

            data => outputData.Add(data),

            new ExecutionDataflowBlockOptions()

            {

                MaxDegreeOfParallelism = 1 // Must be 1 here, as lists are not concurrent collection

            });

        // Link the blocks together

        callBlock.LinkTo(

            aggregateBlock,

            new DataflowLinkOptions()

            {

                PropagateCompletion = true

            });

        return (callBlock, aggregateBlock.Completion, outputData);

    }

    static Task<string> CallApi(int value) => Task.FromResult(value.ToString());

}