Net Fx 4.0 Review

public async Task<WorkItem> ProcessAsync(WorkItem input, CancellationToken token) { if (!input.IsValid) return input;

// Create consumer tasks for each pipeline stage var stageTasks = new List<Task>(); for (int i = 0; i < _stages.Count; i++) { var stageIndex = i; var stage = _stages[stageIndex]; var nextQueue = (stageIndex < _stages.Count - 1) ? new BlockingCollection<WorkItem>() : null; var stageTask = Task.Run(async () => { var sourceQueue = (stageIndex == 0) ? inputQueue : GetQueueForStage(stageIndex - 1); foreach (var item in sourceQueue.GetConsumingEnumerable()) { cancellationToken.ThrowIfCancellationRequested(); progress?.Report($"Processing item {item.Id} in {stage.StageName}"); var processedItem = await stage.ProcessAsync(item, cancellationToken); if (nextQueue != null) { nextQueue.Add(processedItem, cancellationToken); } else { results.Add(processedItem); } } nextQueue?.CompleteAdding(); }, cancellationToken); stageTasks.Add(stageTask); StoreQueueForStage(stageIndex, nextQueue); } await Task.WhenAll(stageTasks.ToArray()); await producerTask; return results.ToList(); } private Dictionary<int, BlockingCollection<WorkItem>> _queues = new Dictionary<int, BlockingCollection<WorkItem>>(); private void StoreQueueForStage(int stageIndex, BlockingCollection<WorkItem> queue) { if (queue != null) _queues[stageIndex] = queue; } private BlockingCollection<WorkItem> GetQueueForStage(int stageIndex) { return _queues.ContainsKey(stageIndex) ? _queues[stageIndex] : null; } } 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); _queues[stageIndex] : null

// Stage 2: Data Transformation public class TransformationStage : IPipelineStage<WorkItem, WorkItem> { public string StageName => "Transformation"; } } public async Task&lt

// Stage 3: Data Enrichment public class EnrichmentStage : IPipelineStage<WorkItem, WorkItem> { public string StageName => "Enrichment";

// Stage 1: Data Validation public class ValidationStage : IPipelineStage<WorkItem, WorkItem> { public string StageName => "Validation";

// Simulate enrichment (e.g., database lookup, API call) await Task.Delay(80, token); input.ProcessedData = $"[ENRICHED] {input.ProcessedData} - Length: {input.ProcessedData.Length}"; Console.WriteLine($"[{StageName}] Item {input.Id}: Enriched data"); return input; } }