func-to-pipeline.cs

void Main() {
  var doWork = CreatePipeLineFunc(async (int x) => {
    var delay = ((x % 3) + 1) * 1000;
    await Task.Delay(delay);
    return new { x, delay };
  }, maxDegreeOfParallelism: 10);

  Enumerable.Range(1, 100).Select(m => doWork(m)).Dump();

}

/// Turns an async function into a function that is serialized by the 
Func<T, Task<U>> CreatePipeLineFunc<T, U>(Func<T, Task<U>> f, int maxDegreeOfParallelism = 1) {
  var pipeLine = new ActionBlock<Task<Task<U>>>(async trigger => {
    trigger.Start();
    var task = await trigger;
    await task;
  }, new ExecutionDataflowBlockOptions {
    BoundedCapacity = maxDegreeOfParallelism,
    MaxDegreeOfParallelism = maxDegreeOfParallelism
  });


  return async x => {
    //  Create the trigger to 
    U result = default(U);
    Task<U> taskToRun = null;
    var trigger = new Task<Task<U>>(() => {
      taskToRun = f(x);
      return taskToRun;
    });
    await pipeLine.SendAsync(trigger);
    await trigger;
    return await taskToRun;
  };
}