using System;

using System.Collections.Generic;

using System.Linq;

using System.Threading;

using System.Threading.Tasks;

using System.Threading.Tasks.Dataflow;

​

public class Program

{

    public static async Task Main()

    {

        var xlsBlock = new TransformBlock<int, int>(document =>

        {

            int duration = 300 + document % 3 * 300;

            Thread.Sleep(duration); // Simulate CPU-bound work

            return document;

        });

        var pdfBlock = new TransformBlock<int, int>(document =>

        {

            int duration = 100 + document % 5 * 200;

            Thread.Sleep(duration); // Simulate CPU-bound work

            return document;

        });

​

        var orderRestorer = CreateRestoreOrderBlock<int>(

            indexSelector: document => document, startingIndex: 1L);

​

        var uploader = new ActionBlock<int>(async document =>

        {

            Console.WriteLine("{0:HH:mm:ss.fff} Uploading document #{1}", DateTime.Now, document);

            await Task.Delay(500); // Simulate I/O-bound work

        });

​

        xlsBlock.LinkTo(orderRestorer);

        pdfBlock.LinkTo(orderRestorer);

        orderRestorer.LinkTo(uploader, new DataflowLinkOptions { PropagateCompletion = true });

​

        foreach (var document in Enumerable.Range(1, 10))

        {

            if (document % 2 == 0)

                xlsBlock.Post(document);

            else

                pdfBlock.Post(document);

        }

        xlsBlock.Complete();

        pdfBlock.Complete();

        _ = Task.WhenAll(xlsBlock.Completion, pdfBlock.Completion)

            .ContinueWith(_ => orderRestorer.Complete());

​

        await uploader.Completion;

    }

​

    /// <summary>Creates a dataflow block that restores the order of

    /// a shuffled pipeline.</summary>

    public static IPropagatorBlock<T, T> CreateRestoreOrderBlock<T>(

        Func<T, long> indexSelector,

        long startingIndex = 0L,

        DataflowBlockOptions options = null)

    {

        if (indexSelector == null) throw new ArgumentNullException(nameof(indexSelector));

        var executionOptions = new ExecutionDataflowBlockOptions();

        if (options != null)

        {

            executionOptions.CancellationToken = options.CancellationToken;

            executionOptions.BoundedCapacity = options.BoundedCapacity;

            executionOptions.EnsureOrdered = options.EnsureOrdered;

            executionOptions.TaskScheduler = options.TaskScheduler;

            executionOptions.MaxMessagesPerTask = options.MaxMessagesPerTask;

            executionOptions.NameFormat = options.NameFormat;

        }

​

        var buffer = new Dictionary<long, T>();

        long minIndex = startingIndex;

​

        IEnumerable<T> Transform(T item)

        {

            // No synchronization needed because MaxDegreeOfParallelism = 1

            long index = indexSelector(item);

            if (index < startingIndex)

                throw new InvalidOperationException($"Index {index} is out of range.");

            if (index < minIndex)

                throw new InvalidOperationException($"Index {index} has been consumed.");

            if (buffer.ContainsKey(index))

                throw new InvalidOperationException($"Index {index} is not unique.");

            buffer.Add(index, item);

            while (buffer.TryGetValue(minIndex, out var minItem))

            {

                buffer.Remove(minIndex);

                minIndex++;

                yield return minItem;

            }

        }

​

        return new TransformManyBlock<T, T>(Transform, executionOptions);

    }

​

}

​