Net Fx 4.0 [ PREMIUM ]

// Pipeline stage interface public interface IPipelineStage<TInput, TOutput> { Task<TOutput> ProcessAsync(TInput input, CancellationToken token); string StageName { get; } }

// Parallel Pipeline Processor public class ParallelPipelineProcessor { private readonly List<IPipelineStage<WorkItem, WorkItem>> _stages; private readonly int _parallelismLevel; net fx 4.0

public async Task<List<WorkItem>> ProcessItemsAsync( IEnumerable<WorkItem> items, CancellationToken cancellationToken, IProgress<string> progress = null) { var inputQueue = new BlockingCollection<WorkItem>(); var results = new ConcurrentBag<WorkItem>(); // Start producer task var producerTask = Task.Run(() => { foreach (var item in items) { cancellationToken.ThrowIfCancellationRequested(); inputQueue.Add(item, cancellationToken); } inputQueue.CompleteAdding(); }, cancellationToken); string StageName { get