using System;

using System.Threading;

using System.Threading.Tasks;

using System.Linq;

using System.Threading.Tasks.Dataflow;

​

public static class Program

{

    static async Task Main(string[] args)

    {

        var producer = CreateDeadTimeTransformBlock<int, string>(async (item, deadTime) =>

        {

            ConsolePrint($"Processing #{item}/A Started");

            await Task.Delay(500); // Simulate some I/O operation

            ConsolePrint($"Processing #{item}/A Finished");

​

            await deadTime.MinimumDelayAsync(TimeSpan.FromSeconds(1));

            //await Task.Delay(TimeSpan.FromSeconds(1)); // Undesirable

​

            ConsolePrint($"Processing #{item}/B Started");

            await Task.Delay(500); // Simulate some I/O operation

            ConsolePrint($"Processing #{item}/B Finished");

            return item.ToString();

        }, new ExecutionDataflowBlockOptions()

        {

            MaxDegreeOfParallelism = 2

        });

​

        var consumer = new ActionBlock<string>(item =>

        {

            ConsolePrint($"Received: {item}");

        });

​

        producer.LinkTo(consumer, new DataflowLinkOptions() { PropagateCompletion = true });

​

        foreach (var item in Enumerable.Range(1, 5))

        {

            await producer.SendAsync(item);

        }

        producer.Complete();

        await consumer.Completion;

        ConsolePrint($"Finished");

    }

​

    public class DeadTime

    {

        private readonly SemaphoreSlim _semaphore;

        private readonly CancellationToken _cancellationToken;

​

        public DeadTime(SemaphoreSlim semaphore, CancellationToken cancellationToken)

        {

            _semaphore = semaphore;

            _cancellationToken = cancellationToken;

        }

​

        public async Task MinimumDelayAsync(TimeSpan timeSpan)

        {

            _semaphore.Release();

            try

            {

                await Task.Delay(timeSpan, _cancellationToken);

            }

            finally

            {

                await _semaphore.WaitAsync();

            }

        }

    }

​

    public static TransformBlock<TInput, TOutput>

        CreateDeadTimeTransformBlock<TInput, TOutput>(

        Func<TInput, DeadTime, Task<TOutput>> transform,

        ExecutionDataflowBlockOptions dataflowBlockOptions)

    {

        var cancellationToken = dataflowBlockOptions.CancellationToken;

        var maxDOP = dataflowBlockOptions.MaxDegreeOfParallelism;

        var taskScheduler = dataflowBlockOptions.TaskScheduler;

​

        SemaphoreSlim semaphore;

        if (maxDOP == DataflowBlockOptions.Unbounded)

        {

            // This is not an interesting case

            semaphore = new SemaphoreSlim(Int32.MaxValue);

        }

        else

        {

            semaphore = new SemaphoreSlim(maxDOP, maxDOP);

​

            // The degree of parallelism is controlled by the semaphore

            dataflowBlockOptions.MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded;

​

            // Use a limited-concurrency scheduler for preserving the processing order

            dataflowBlockOptions.TaskScheduler = new ConcurrentExclusiveSchedulerPair(

                taskScheduler, maxDOP).ConcurrentScheduler;

        }

​

        var deadTime = new DeadTime(semaphore, cancellationToken);

​

        var block = new TransformBlock<TInput, TOutput>(async item =>

        {

            await semaphore.WaitAsync(cancellationToken);

            try

            {

                return await transform(item, deadTime);

            }

            finally

            {

                semaphore.Release();

            }

        }, dataflowBlockOptions);

​

        dataflowBlockOptions.MaxDegreeOfParallelism = maxDOP; // Restore initial value

        dataflowBlockOptions.TaskScheduler = taskScheduler; // Restore initial value

        return block;

    }

​

    public static void ConsolePrint(string line)

    {

        Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff} [{Thread.CurrentThread.ManagedThreadId}] > {line}");

    }

}

​