Philiprehberger.BatchProcessor 0.4.0

Philiprehberger.BatchProcessor

Process large collections in configurable batches with progress reporting, error handling, async execution, streaming IAsyncEnumerable support, checkpoint/resume, and adaptive batch sizing.

Installation

dotnet add package Philiprehberger.BatchProcessor

Usage

using Philiprehberger.BatchProcessor;

var items = Enumerable.Range(1, 1000).ToList();

var result = await BatchProcessor.Process(items, batchSize: 100, async batch =>
{
    await ProcessItemsAsync(batch);
});

Console.WriteLine($"Processed {result.SuccessCount} items in {result.TotalDuration.TotalSeconds}s");

Progress Reporting

using Philiprehberger.BatchProcessor;

var items = Enumerable.Range(1, 500).ToList();

var result = await BatchProcessor.Process(items, batchSize: 50, async batch =>
{
    await SendToApiAsync(batch);
}, new BatchOptions
{
    OnProgress = progress =>
    {
        Console.WriteLine($"Progress: {progress.Percent:F1}% ({progress.ProcessedCount}/{progress.TotalCount})");
    }
});

Parallel Execution with Error Handling

using Philiprehberger.BatchProcessor;

var items = Enumerable.Range(1, 1000).ToList();

var result = await BatchProcessor.Process(items, batchSize: 50, async batch =>
{
    await SendToApiAsync(batch);
}, new BatchOptions
{
    MaxDegreeOfParallelism = 4,
    OnBatchError = BatchErrorHandling.Skip,
    RetryCount = 2
});

Console.WriteLine($"Success: {result.SuccessCount}, Failures: {result.FailureCount}");
Console.WriteLine($"Errors: {result.Errors.Count}");

Per-Batch Timeout

using Philiprehberger.BatchProcessor;

var items = Enumerable.Range(1, 1000).ToList();

var result = await BatchProcessor.Process(items, batchSize: 100, async batch =>
{
    await SendToSlowServiceAsync(batch);
}, new BatchOptions
{
    BatchTimeout = TimeSpan.FromSeconds(30),
    OnBatchError = BatchErrorHandling.Skip
});

Per-Item Error Tracking

using Philiprehberger.BatchProcessor;

var items = Enumerable.Range(1, 100).ToList();

var result = await BatchProcessor.ProcessAsync(items, batchSize: 10, async batch =>
{
    await ProcessBatchAsync(batch);
});

Console.WriteLine($"Succeeded: {result.SucceededCount}, Failed: {result.FailedCount}");

foreach (var failure in result.Failures)
{
    Console.WriteLine($"Item {failure.Item} failed: {failure.Exception?.Message}");
}

Batch Completed Callback

using Philiprehberger.BatchProcessor;

var items = Enumerable.Range(1, 500).ToList();

var result = await BatchProcessor.Process(items, batchSize: 50, async batch =>
{
    await ProcessBatchAsync(batch);
}, new BatchOptions
{
    OnBatchCompleted = e =>
    {
        Console.WriteLine($"Batch {e.BatchIndex}: {e.SuccessCount} ok, {e.FailureCount} failed in {e.Elapsed.TotalMilliseconds}ms");
    }
});

Cancellation Token

using Philiprehberger.BatchProcessor;

using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5));
var items = Enumerable.Range(1, 10000).ToList();

var result = await BatchProcessor.Process(items, batchSize: 100, async batch =>
{
    await ProcessBatchAsync(batch);
}, cancellationToken: cts.Token);

Streaming with IAsyncEnumerable

Process items from an async stream without materializing the full collection:

using Philiprehberger.BatchProcessor;

async IAsyncEnumerable<Order> GetOrdersAsync()
{
    await foreach (var order in database.StreamOrdersAsync())
    {
        yield return order;
    }
}

var result = await BatchProcessor.ProcessStreamAsync(GetOrdersAsync(), batchSize: 50, async batch =>
{
    await SendToWarehouseAsync(batch);
}, new BatchOptions
{
    MaxDegreeOfParallelism = 3,
    OnBatchError = BatchErrorHandling.Skip
});

Console.WriteLine($"Streamed {result.SuccessCount} orders, {result.FailureCount} failed");

Checkpoint and Resume

Persist progress after each batch and resume from where processing left off:

using Philiprehberger.BatchProcessor;

var lastCheckpoint = LoadCheckpointFromDisk(); // e.g., returns 5

var result = await BatchProcessor.Process(items, batchSize: 100, async batch =>
{
    await ProcessBatchAsync(batch);
}, new BatchOptions
{
    ResumeFromBatch = lastCheckpoint + 1,
    CheckpointCallback = batchIndex =>
    {
        SaveCheckpointToDisk(batchIndex);
    }
});

Streaming with Per-Item Results

When you need per-item failure tracking on a streamed source, use ProcessStreamWithItemsAsync:

using Philiprehberger.BatchProcessor;

async IAsyncEnumerable<Order> GetOrdersAsync()
{
    await foreach (var order in database.StreamOrdersAsync())
    {
        yield return order;
    }
}

var result = await BatchProcessor.ProcessStreamWithItemsAsync(GetOrdersAsync(), batchSize: 50, async batch =>
{
    await SendToWarehouseAsync(batch);
}, new BatchOptions
{
    OnBatchError = BatchErrorHandling.Skip
});

foreach (var failure in result.Failures)
{
    Console.WriteLine($"Order {failure.Item} failed: {failure.Exception?.Message}");
}

Adaptive Batch Sizing

Automatically adjust batch sizes based on measured throughput:

using Philiprehberger.BatchProcessor;

var items = Enumerable.Range(1, 10000).ToList();

var result = await BatchProcessor.Process(items, batchSize: 50, async batch =>
{
    await SendToApiAsync(batch);
}, new BatchOptions
{
    AdaptiveBatching = new AdaptiveBatchOptions
    {
        MinBatchSize = 10,
        MaxBatchSize = 500,
        TargetThroughput = 200 // items per second
    },
    OnBatchCompleted = e =>
    {
        Console.WriteLine($"Batch {e.BatchIndex}: {e.ItemCount} items in {e.Elapsed.TotalMilliseconds}ms");
    }
});

API

BatchProcessor

BatchOptions

AdaptiveBatchOptions

BatchProgress

BatchResult

BatchResult<T>

BatchItemResult<T>

BatchCompletedEventArgs

BatchError

Development

dotnet build src/Philiprehberger.BatchProcessor.csproj --configuration Release

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT