FluentParallel 0.2.0

FluentParallel

A fluent, composable API for running parallel and streaming data pipelines in .NET.
It allows you to easily batch, transform, filter, and stream results as they complete using a simple fluent interface.

✨ Features

• Fluent API to build parallel pipelines step by step.
• Batching of items into configurable chunk sizes.
• Synchronous, asynchronous, and streaming transformations.
• Filtering at any stage in the pipeline.
• Error handling with OnError.
• Works with both IEnumerable<T> and IAsyncEnumerable<T> sources.
• Real-time streaming with ExecuteStreamAsync + StreamAsync.

🚀 Installation

dotnet add package FluentParallel

🚀 Before vs After - See the Difference

❌ "Raw" C# (manual batching + concurrency + channel)

// BEFORE: raw C# with batching, parallelism and channel
async Task RunRawExampleAsync()
{
    var customerIds = Enumerable.Range(1, 1000);
    int batchSize = 10;

    var batches = customerIds
        .Select((id, index) => new { id, index })
        .GroupBy(x => x.index / batchSize)
        .Select(g => g.Select(x => x.id).ToArray())
        .ToList();

    var semaphore = new SemaphoreSlim(4);
    var channel = Channel.CreateUnbounded<string>();
    var cts = new CancellationTokenSource();

    // Producer
    var producer = Task.Run(async () =>
    {
        foreach (var batch in batches)
        {
            await semaphore.WaitAsync(cts.Token);
            _ = Task.Run(async () =>
            {
                try
                {
                    // Simulate async work
                    await Task.Delay(500);
                    var sum = batch.Sum();
                    await channel.Writer.WriteAsync($"Batch sum: {sum}", cts.Token);
                }
                finally
                {
                    semaphore.Release();
                }
            }, cts.Token);
        }

        // Wait for all batches to finish then complete channel
        for (int i = 0; i < 4; i++)
            await semaphore.WaitAsync(cts.Token);

        channel.Writer.Complete();
    }, cts.Token);

    // Consumer
    await foreach (var msg in channel.Reader.ReadAllAsync(cts.Token))
    {
        Console.WriteLine(msg);
    }

    await producer;
}

✅ After with FluentParallel

// AFTER: FluentParallel handles batching, parallelism and streaming for you!
async Task RunFluentParallelExampleAsync()
{
    var customerIds = Enumerable.Range(1, 1000);

    await foreach (var sum in ParallelFlow
        .From(customerIds)
        .Batch(10)                     // auto-batching
        .WithParallel(4)               // 4 batches in parallel
        .ExecuteStreamAsync(async batch =>
        {
            // Simulate async work
            await Task.Delay(500);
            return batch.Sum();
        })
        .StreamAsync())                // stream results as they complete
    {
        Console.WriteLine($"Batch sum: {sum}");
    }
}

✨ The difference

• Before: ~45 lines of boilerplate with SemaphoreSlim, Channel, manual batching and tasks.
• After: ~10 lines, all in one fluent pipeline.
• You get parallelism, batching, streaming and error handling built-in.

🧩 When to Use

Use FluentParallel when you need to:

• Process large datasets without waiting for all items to complete.
• Combine async tasks and parallelism seamlessly.
• Stream intermediate results as they are ready.
• Handle errors gracefully without stopping the pipeline.

How to use it?

1️⃣ Synchronous Flow

using FluentParallel;

var numbers = Enumerable.Range(1, 30);

var result = await ParallelFlow
    .From(numbers)
    .Batch(5)          // batches of 5
    .WithParallel(4)   // 4 batches processed in parallel
    .Execute(batches =>
    {
        Console.WriteLine(
            $"Thread {Environment.CurrentManagedThreadId}: [{string.Join(",", batches)}]");
        return batches;
    })
    .CollectAsync();

Console.WriteLine($"Collected {result.Count} batches");

2️⃣ Asynchronous Flow

var resultAsync = await ParallelFlow
    .From(numbers)
    .Batch(5)
    .WithParallel(4)
    .ExecuteAsync(async batch =>
    {
        // Simulate async work
        await Task.Delay(2000);
        Console.WriteLine(
            $"Thread {Environment.CurrentManagedThreadId}: [{string.Join(",", batch)}] | sum: {batch.Sum()}");
        return batch.Sum();
    })
    .Filter(sum => sum < 100) // filter items < 100
    .CollectAsync();

Console.WriteLine($"Collected sums: [{string.Join(",", resultAsync)}]");

3️⃣ Streaming Flow (process results as soon as they are ready!)

await foreach (var result in ParallelFlow
    .From(numbers)
    .Batch(5)
    .WithParallel(4)
    .ExecuteStreamAsync(async batch =>
    {
        await Task.Delay(2000);
        Console.WriteLine(
            $"Thread {Environment.CurrentManagedThreadId}: [{string.Join(",", batch)}] | sum: {batch.Sum()}");
        return batch.Sum();
    })
    .Filter(sum => sum < 100)
    .StreamAsync()) // yields as results come in
{
    Console.WriteLine($"Got: {result}");
}

4️⃣ Error Handling Flow

This demonstrates that:

• Exceptions inside steps are caught by your OnError handler.
• Processing continues for other items even if one fails.

await foreach (var result in ParallelFlow
    .From(numbers)
    .Batch(5)
    .WithParallel(4)
    .OnError(ex => Console.WriteLine($"Error: {ex.Message}"))
    .ExecuteStreamAsync(async n =>
    {
        await Task.Delay(500);
        Console.WriteLine($"Thread {Environment.CurrentManagedThreadId}: [{string.Join(",", n)}]");

        if (n.Contains(5))
            throw new InvalidOperationException($"Boom! Thread {Environment.CurrentManagedThreadId} error: contains 5!");
        return n.Sum();
    })
    .StreamAsync())
{
    Console.WriteLine($"Sum: {result}");
}

📄 License

MIT