Snail.Toolkit.Async.Pipeline 1.0.0

Snail.Toolkit.Async.Pipeline

AsyncPipeline<T> is a lightweight, fluent wrapper around IAsyncEnumerable<T>. It simplifies asynchronous stream processing in C# by providing LINQ-like operators, parallel execution capabilities, and seamless integration with standard collection types.

The SelectMany Operator

The SelectMany operator is used to flatten "nested" sequences. It projects each element of the source stream into a sub-sequence and flattens those sub-sequences into a single, continuous asynchronous stream.

1. Synchronous Flattening

Use this when each item in your async stream contains a standard collection (like List<T> or Array) that you want to expand.

// Example: Extracting all tags from a stream of Posts
IAsyncEnumerable<string> tagStream = postPipeline
    .SelectMany(post => post.Tags); // post.Tags is a List<string>

2. Asynchronous Flattening

Use this when the transformation itself returns another IAsyncEnumerable<T>. This is common when querying child records from a database or an API for every item in the source.

// Example: Fetching comments for every user in the pipeline
var commentsPipeline = userPipeline
    .SelectMany(user => commentService.GetCommentsAsync(user.Id));

3. Parallel Flattening

This overload allows you to process multiple source items concurrently. It triggers multiple sub-sequences at once and interleaves their results into the main pipeline as they arrive.

// Example: Scanning multiple directories for files in parallel
var filePipeline = directories
    .ToPipeline()
    .SelectMany(async (path, ct) => fileService.SearchFilesAsync(path, ct), 
                maxDegreeOfParallelism: 4);

Basic Usage & Entry Points

Creating a Pipeline

You can initialize a pipeline using the AsyncPipelineBuilder or the provided extension methods:

// From an IAsyncEnumerable
var pipe1 = myAsyncSource.ToPipeline();

// From a synchronous IEnumerable
var pipe2 = myList.ToPipeline();

// From a Task/ValueTask
var pipe3 = AsyncPipelineBuilder.From(GetSingleItemAsync());

Common Operations

The pipeline supports standard LINQ operations adapted for asynchrony:

Terminal Methods

These methods execute the pipeline and produce a result:

// Collect into a List
List<User> users = await pipeline.ToListAsync();

// Get the first item
User? first = await pipeline.FirstOrDefaultAsync();

// Aggregate values
int total = await pipeline.AggregateAsync(0, (sum, val) => sum + val);

Why use AsyncPipeline?

• Readable Syntax: Avoids deep nesting of await foreach loops.
• Performance: Built with ValueTask and System.Threading.Channels to ensure high throughput with low allocation overhead.
• Parallelism Made Easy: Simplifies complex SemaphoreSlim logic into a single maxDegreeOfParallelism parameter.
• Cancellation Support: First-class support for CancellationToken across all operators.

License

Snail.Toolkit.Async.Pipeline is a free and open source project, released under the permissible MIT license.