Musoftware.BatchProcessor 1.1.0

Musoftware.BatchProcessor

High-performance, domain-agnostic generic batch processing library for .NET with rate limiting, real-time result streaming, exponential retries, circuit breaker protection, dead letter queue support, sliding-window speed monitoring, and auto-slowdown detection.

Features

• ⚡ High Performance & Parallel Execution: Configurable MaxDegreeOfParallelism and BatchSize.
• 🛑 Rate Limiting & Throughput Cap: Enforce MaxItemsPerSecond to prevent API rate limiting (429 errors).
• 🌊 Real-Time Result Streaming: Process results as they finish via ProcessStreamingAsync (IAsyncEnumerable<BatchItemResult<TInput, TOutput>>).
• 🔄 Advanced Retry Strategies: Support for Fixed, Exponential, and ExponentialWithJitter backoffs with exception filtering predicates.
• ⚡ Circuit Breaker Protection: Pause execution when consecutive failure thresholds are breached (WithCircuitBreaker).
• 📬 Dead Letter Queue (DLQ): Automatic capture and export of failed items via IDeadLetterCollector<TInput, TOutput>.
• 📊 Sliding-Window Speed Monitor: Real-time throughput (items/sec) tracking and progress metrics (BatchProgress).
• ⚠️ Auto-Slowdown Detection: Trigger events (OnSlowdownDetected) when throughput drops below minimum thresholds.
• 🎯 Domain Agnostic: Fully generic IBatchProcessor<TInput, TOutput> for any data type or processing delegate.
• ⚙️ Fluent Configuration: Fluent BatchProcessorOptionsBuilder for clean initialization.

Installation

dotnet add package Musoftware.BatchProcessor

Quick Start

using Musoftware.BatchProcessor;

// 1. Configure options with Fluent Builder
var options = new BatchProcessorOptionsBuilder()
    .WithMaxDegreeOfParallelism(16)
    .WithBatchSize(1000)
    .WithMaxItemsPerSecond(500.0) // Rate limiting
    .WithRetry(
        retryCount: 3,
        retryDelay: TimeSpan.FromMilliseconds(100),
        strategy: RetryBackoffStrategy.ExponentialWithJitter,
        shouldRetry: ex => ex is HttpRequestException || ex is TimeoutException)
    .WithCircuitBreaker(failureThreshold: 10, breakDuration: TimeSpan.FromSeconds(5))
    .WithContinueOnError(true)
    .Build();

var engine = new BatchProcessingEngine<string, int>(options);

// 2. Stream Batch Results in Real Time
var items = Enumerable.Range(1, 10000).Select(i => $"Item_{i}");

await foreach (var result in engine.ProcessStreamingAsync(
    items,
    async (item, cancellationToken) =>
    {
        await Task.Delay(10, cancellationToken);
        return item.Length;
    }))
{
    if (result.IsSuccess)
    {
        Console.WriteLine($"Processed {result.Input} -> {result.Output}");
    }
}

License

MIT