🚀 TaskListProcessor

A modern, enterprise-grade .NET 9.0 library for orchestrating asynchronous operations with comprehensive telemetry, circuit breakers, dependency injection, and advanced scheduling capabilities.

📑 Table of Contents

🚀 TaskListProcessor

✨ Overview

TaskListProcessor is a production-ready .NET 9.0 library designed to solve complex asynchronous orchestration challenges in modern applications. Built with enterprise-grade patterns including dependency injection, circuit breakers, task scheduling, and comprehensive telemetry, it provides a robust foundation for high-throughput, fault-tolerant systems.

🎯 Why TaskListProcessor?

The Problem: Modern applications require sophisticated coordination of multiple async operations—API calls, database queries, file I/O, microservice interactions—while maintaining resilience, observability, and performance under varying loads.

The Solution: TaskListProcessor provides a battle-tested, enterprise-ready framework with:

🛡️ Fault Isolation: Circuit breakers and individual task failure isolation
📊 Enterprise Observability: OpenTelemetry integration with rich metrics and tracing
⚡ Advanced Scheduling: Priority-based, dependency-aware task execution
🎯 Type Safety: Strongly-typed results with comprehensive error categorization
🔧 Dependency Injection: Native .NET DI integration with decorator pattern support
🏗️ Interface Segregation: Clean, focused interfaces following SOLID principles

🔥 Key Features

Core Processing Capabilities

🚀 Concurrent Execution: Parallel task processing with configurable concurrency limits and load balancing
🛡️ Circuit Breaker Pattern: Automatic failure detection and cascading failure prevention
📊 Rich Telemetry: Comprehensive timing, success rates, error tracking, and OpenTelemetry integration
🎯 Type Safety: Strongly-typed results with full IntelliSense support and error categorization
⏱️ Timeout & Cancellation: Built-in support for graceful shutdown and per-task timeouts
🔄 Task Dependencies: Dependency resolution with topological sorting and execution ordering

Enterprise Architecture Features

🏗️ Dependency Injection: Native .NET DI integration with fluent configuration API
🎨 Interface Segregation: Clean, focused interfaces following SOLID principles
� Decorator Pattern: Pluggable cross-cutting concerns (logging, metrics, circuit breakers)
📈 Advanced Scheduling: Priority-based, FIFO, LIFO, and custom scheduling strategies
🧵 Thread Safety: Lock-free concurrent collections and thread-safe operations
💾 Memory Optimization: Object pooling and efficient memory management

Developer Experience

� Structured Logging: Integration with Microsoft.Extensions.Logging and Serilog
🔍 Health Checks: Built-in health monitoring and diagnostic capabilities
� Streaming Results: Async enumerable support for real-time result processing
🧪 Testing Support: Comprehensive test helpers and mock-friendly interfaces
📖 Rich Documentation: Extensive XML documentation and practical examples

🏗️ Architecture

TaskListProcessor implements a modern, enterprise-ready architecture with clear separation of concerns:

┌─────────────────────────────────────────────────────────────────┐
│                    Dependency Injection Layer                  │
│        services.AddTaskListProcessor().WithAllDecorators()     │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                      Decorator Chain                            │
│  LoggingDecorator → MetricsDecorator → CircuitBreakerDecorator  │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Interface Segregation Layer                 │
│  ITaskProcessor │ ITaskBatchProcessor │ ITaskStreamProcessor   │
│              ITaskTelemetryProvider                           │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Core Processing Engine                       │
│            TaskListProcessorEnhanced (Backward Compatible)     │
└─────────────────────────────────────────────────────────────────┘
                                │
              ┌─────────────────┼─────────────────┐
              │                 │                 │
    ┌─────────▼────────┐ ┌─────▼──────┐ ┌───────▼──────┐
    │ TaskDefinition   │ │TaskTelemetry│ │TaskProgress  │
    │ + Dependencies   │ │ + Metrics   │ │ + Reporting  │
    │ + Priority       │ │ + Tracing   │ │ + Streaming  │
    │ + Scheduling     │ │ + Health    │ │ + Estimates  │
    └──────────────────┘ └────────────┘ └──────────────┘

Core Components

Interface Layer: Clean, focused interfaces for different processing scenarios
Decorator Layer: Cross-cutting concerns (logging, metrics, circuit breakers)
Processing Engine: Thread-safe orchestration with advanced scheduling
Telemetry System: Comprehensive observability and health monitoring
Dependency Resolution: Topological sorting and execution ordering
Circuit Breaker: Cascading failure prevention and automatic recovery

⚡ Quick Start

Installation

# Clone the repository
git clone https://github.com/markhazleton/TaskListProcessor.git
cd TaskListProcessor

# Build the solution
dotnet build

# Run the demo
dotnet run --project examples/TaskListProcessor.Console

Basic Usage (Direct Instantiation)

using TaskListProcessing.Core;
using Microsoft.Extensions.Logging;

// Set up logging (optional but recommended)
using var loggerFactory = LoggerFactory.Create(builder => builder.AddConsole());
var logger = loggerFactory.CreateLogger<Program>();

// Create the processor
using var processor = new TaskListProcessorEnhanced("My Tasks", logger);

// Define your tasks using the factory pattern
var taskFactories = new Dictionary<string, Func<CancellationToken, Task<object?>>>
{
    ["Weather Data"] = async ct => await GetWeatherAsync("London"),
    ["Stock Prices"] = async ct => await GetStockPricesAsync("MSFT"),
    ["User Data"] = async ct => await GetUserDataAsync(userId)
};

// Execute all tasks concurrently
await processor.ProcessTasksAsync(taskFactories, cancellationToken);

// Access results and telemetry
foreach (var result in processor.TaskResults)
{
    Console.WriteLine($"{result.Name}: {(result.IsSuccessful ? "✅" : "❌")}");
}

Dependency Injection Usage (Recommended)

using TaskListProcessing.Extensions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

// Program.cs
var builder = Host.CreateApplicationBuilder(args);

// Configure TaskListProcessor with decorators
builder.Services.AddTaskListProcessor(options =>
{
    options.MaxConcurrentTasks = 10;
    options.EnableDetailedTelemetry = true;
    options.CircuitBreakerOptions = new() { FailureThreshold = 3 };
})
.WithLogging()
.WithMetrics()
.WithCircuitBreaker();

var host = builder.Build();

// Usage in your services
public class MyService
{
    private readonly ITaskBatchProcessor _processor;
    
    public MyService(ITaskBatchProcessor processor)
    {
        _processor = processor;
    }
    
    public async Task ProcessDataAsync()
    {
        var tasks = new Dictionary<string, Func<CancellationToken, Task<object?>>>
        {
            ["API Call"] = async ct => await CallApiAsync(ct),
            ["DB Query"] = async ct => await QueryDatabaseAsync(ct)
        };
        
        await _processor.ProcessTasksAsync(tasks);
    }
}

📖 Comprehensive Examples

Travel Dashboard (Real-world Scenario)

This example demonstrates fetching weather and activities data for multiple cities:

using var processor = new TaskListProcessorEnhanced("Travel Dashboard", logger);
using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));

var cities = new[] { "London", "Paris", "Tokyo", "New York" };
var taskFactories = new Dictionary<string, Func<CancellationToken, Task<object?>>>();

// Create tasks for each city
foreach (var city in cities)
{
    taskFactories[$"{city} Weather"] = ct => weatherService.GetWeatherAsync(city, ct);
    taskFactories[$"{city} Activities"] = ct => activitiesService.GetActivitiesAsync(city, ct);
}

// Execute and handle results
try 
{
    await processor.ProcessTasksAsync(taskFactories, cts.Token);
    
    // Group results by city
    var cityData = processor.TaskResults
        .GroupBy(r => r.Name.Split(' ')[0])
        .ToDictionary(g => g.Key, g => g.ToList());
    
    // Display results with rich formatting
    foreach (var (city, results) in cityData)
    {
        Console.WriteLine($"\n🌍 {city}:");
        foreach (var result in results)
        {
            var status = result.IsSuccessful ? "✅" : "❌";
            Console.WriteLine($"  {status} {result.Name.Split(' ')[1]}");
        }
    }
}
catch (OperationCanceledException)
{
    logger.LogWarning("Operation timed out after 2 minutes");
}

Individual Task Execution

For scenarios requiring fine-grained control:

// Execute a single task with full telemetry
var result = await processor.ExecuteTaskAsync(
    "Critical API Call", 
    httpClient.GetStringAsync("https://api.example.com/data"),
    cancellationToken);

if (result.IsSuccessful)
{
    var data = result.Data;
    logger.LogInformation("Received {DataLength} characters", data?.Length ?? 0);
}
else
{
    logger.LogError("API call failed: {Error}", result.ErrorMessage);
}

🔧 Advanced Usage

Task Dependencies & Scheduling

using TaskListProcessing.Models;
using TaskListProcessing.Scheduling;

// Configure with dependency resolution
var options = new TaskListProcessorOptions
{
    DependencyResolver = new TopologicalTaskDependencyResolver(),
    SchedulingStrategy = TaskSchedulingStrategy.Priority,
    MaxConcurrentTasks = Environment.ProcessorCount * 2
};

using var processor = new TaskListProcessorEnhanced("Advanced Tasks", logger, options);

// Define tasks with dependencies and priorities
var taskDefinitions = new[]
{
    new TaskDefinition
    {
        Name = "Initialize",
        Factory = async ct => await InitializeAsync(ct),
        Priority = TaskPriority.High
    },
    new TaskDefinition
    {
        Name = "Process Data",
        Factory = async ct => await ProcessDataAsync(ct),
        Dependencies = new[] { "Initialize" },
        Priority = TaskPriority.Medium
    },
    new TaskDefinition
    {
        Name = "Generate Report",
        Factory = async ct => await GenerateReportAsync(ct),
        Dependencies = new[] { "Process Data" },
        Priority = TaskPriority.Low
    }
};

await processor.ProcessTaskDefinitionsAsync(taskDefinitions);

Circuit Breaker Configuration

var options = new TaskListProcessorOptions
{
    CircuitBreakerOptions = new CircuitBreakerOptions
    {
        FailureThreshold = 5,
        RecoveryTimeout = TimeSpan.FromMinutes(2),
        MinimumThroughput = 10
    }
};

using var processor = new TaskListProcessorEnhanced("Resilient Tasks", logger, options);

// Tasks will automatically trigger circuit breaker on repeated failures
var taskFactories = new Dictionary<string, Func<CancellationToken, Task<object?>>>
{
    ["Resilient API"] = async ct => await CallExternalApiAsync(ct),
    ["Fallback Service"] = async ct => await CallFallbackServiceAsync(ct)
};

await processor.ProcessTasksAsync(taskFactories);

// Check circuit breaker status
var cbStats = processor.CircuitBreakerStats;
if (cbStats?.State == CircuitBreakerState.Open)
{
    Console.WriteLine($"Circuit breaker opened at {cbStats.OpenedAt}");
}

Streaming Results

using TaskListProcessing.Interfaces;

// Inject the stream processor
public class StreamingService
{
    private readonly ITaskStreamProcessor _streamProcessor;
    
    public StreamingService(ITaskStreamProcessor streamProcessor)
    {
        _streamProcessor = streamProcessor;
    }
    
    public async Task ProcessWithStreamingAsync()
    {
        var tasks = CreateLongRunningTasks();
        
        // Process results as they complete
        await foreach (var result in _streamProcessor.ProcessTasksStreamAsync(tasks))
        {
            Console.WriteLine($"Completed: {result.Name} - {result.IsSuccessful}");
            
            // Process result immediately without waiting for all tasks
            await HandleResultAsync(result);
        }
    }
}

Health Monitoring

var options = new TaskListProcessorOptions
{
    HealthCheckOptions = new HealthCheckOptions
    {
        MinSuccessRate = 0.8, // 80% success rate threshold
        MaxAverageExecutionTime = TimeSpan.FromSeconds(5),
        IncludeCircuitBreakerState = true
    }
};

using var processor = new TaskListProcessorEnhanced("Health Monitored", logger, options);

// After processing tasks
var healthResult = processor.PerformHealthCheck();
if (!healthResult.IsHealthy)
{
    Console.WriteLine($"Health check failed: {healthResult.Message}");
}

// Get detailed telemetry
var telemetrySummary = processor.GetTelemetrySummary();
Console.WriteLine($"Success rate: {telemetrySummary.SuccessRate:F1}%");
Console.WriteLine($"Average execution time: {telemetrySummary.AverageExecutionTime:F0}ms");

📊 Performance & Telemetry

TaskListProcessor provides comprehensive telemetry out of the box:

Built-in Metrics

// Access telemetry after execution
var telemetry = processor.Telemetry;
var successRate = telemetry.Count(t => t.IsSuccessful) / (double)telemetry.Count * 100;
var averageTime = telemetry.Average(t => t.DurationMs);
var throughput = telemetry.Count / telemetry.Max(t => t.DurationMs) * 1000;

Console.WriteLine($"📊 Success Rate: {successRate:F1}%");
Console.WriteLine($"⏱️ Average Time: {averageTime:F0}ms");
Console.WriteLine($"🚀 Throughput: {throughput:F1} tasks/second");

Sample Telemetry Output

=== 📊 TELEMETRY SUMMARY ===
📈 Total Tasks: 16
✅ Successful: 13 (81.2%)
❌ Failed: 3
⏱️ Average Time: 1,305ms
🏃 Fastest: 157ms | 🐌 Slowest: 2,841ms
⏰ Total Execution Time: 20,884ms

=== 📋 DETAILED TELEMETRY ===
✅ Successful Tasks (sorted by execution time):
  🚀 London Things To Do: 157ms
  🚀 Dallas Things To Do: 339ms
  ⚡ Chicago Things To Do: 557ms
  🏃 London Weather: 1,242ms
  ...

❌ Failed Tasks:
  💥 Sydney Things To Do: ArgumentException after 807ms
  💥 Tokyo Things To Do: ArgumentException after 424ms

🛠️ API Reference

Core Interfaces

Interface	Description	Key Methods
`ITaskProcessor`	Single task execution	`ExecuteTaskAsync<T>()`
`ITaskBatchProcessor`	Batch processing	`ProcessTasksAsync()`, `ProcessTaskDefinitionsAsync()`
`ITaskStreamProcessor`	Streaming results	`ProcessTasksStreamAsync()`
`ITaskTelemetryProvider`	Telemetry & health	`GetTelemetrySummary()`, `PerformHealthCheck()`