Open.ChannelExtensions

You can define expressive, asynchronous processing pipelines by chaining extensions on a System.Threading.Channels.Channel. This allows you to source data, apply transformations with controlled concurrency and capacity, and consume the final results in a single fluent statement.

Common pipeline steps include:

• SourceAsync: Populates a channel from an IEnumerable<Task<T>>.
• PipeAsync: Performs asynchronous transformations with maxConcurrency and capacity settings.
• Pipe: Performs synchronous transformations with capacity settings.
• ReadAllAsync: Consumes the final values from the pipeline.

await Channel
    .CreateBounded<T>(10)
    .SourceAsync(source /* IEnumerable<Task<T>> */)
    .PipeAsync(
        maxConcurrency: 2,
        capacity: 5,
        transform: asyncTransform01)
    .Pipe(transform02, /* capacity */ 3)
    .ReadAllAsync(finalTransformedValue => {
        // Do something async with each final value.
    });

The Pipe method allows you to transform values from a source channel into a new channel. You can control the capacity (bounding) and the concurrency of the transformation.

• Unbounded: channel.Pipe(async value => ...)
• Bounded (Capacity): channel.Pipe(async value => ..., capacity)
• Bounded with Concurrency: channel.Pipe(maxConcurrency, async value => ..., capacity)

// Transform to new unbounded channel
var transformed = channel.Pipe(async value => /* transformation */);

// Transform to new unbounded channel with max concurrency X
var transformed = channel.Pipe(X, async value => /* transformation */);

// Transform to new bounded channel with capacity N
var transformed = channel.Pipe(async value => /* transformation */, N);

// Transform to new bounded channel with capacity N and max concurrency X
var transformed = channel.Pipe(X, async value => /* transformation */, N);

// Using named arguments
var transformed = channel.Pipe(
    maxConcurrency: X,
    capacity: N,
    transform: async value => /* transformation */
);

Open.ChannelExtensions provides several ways to consume entries from a channel until it is closed. You can read entries sequentially (one by one) or concurrently.

Sequential Reading

Use ReadAll for synchronous processing or ReadAllAsync for asynchronous processing. Both support providing the entry alone or the entry along with its index.

Concurrent Reading

Use ReadAllConcurrently or ReadAllConcurrentlyAsync to process multiple entries from the channel simultaneously. You must specify a maxConcurrency value.

| Method | Processing Type | Supports Index |

// One by one (Sync)
await channel.ReadAll(entry => { /* ... */ });
await channel.ReadAll((entry, index) => { /* ... */ });

// One by one (Async)
await channel.ReadAllAsync(async entry => { await /* ... */ });
await channel.ReadAllAsync(async (entry, index) => { await /* ... */ });

// Concurrent (Sync)
await channel.ReadAllConcurrently(maxConcurrency, entry => { /* ... */ });

// Concurrent (Async)
await channel.ReadAllConcurrentlyAsync(maxConcurrency, async entry => { await /* ... */ });

The Pipe and PipeAsync methods allow you to create a new channel by transforming the values from an existing channel. This is useful for creating processing stages in a pipeline.

Configuration Options

• Transformation: A function that maps the input value to a new value.
• maxConcurrency: Limits how many transformations are running at the same time. (Available in PipeAsync).
• capacity: Defines the bound of the resulting channel (the number of entries it can hold before applying backpressure).

Usage Patterns

• Unbounded Transformation: channel.Pipe(async value => ...) creates a new unbounded channel.
• Bounded Transformation: channel.Pipe(async value => ..., N) creates a new channel with capacity N.
• Controlled Concurrency: channel.Pipe(X, async value => ..., N) creates a channel with capacity N and a maximum concurrency of X.

// Unbounded
var transformed = channel.Pipe(async value => /* transformation */);

// Bounded with max concurrency X
const X = 4;
var transformed = channel.Pipe(X, async value => /* transformation */);

// Bounded with capacity N
const N = 5;
var transformed = channel.Pipe(async value => /* transformation */, N);

// Bounded with capacity N and max concurrency X
const X = 4;
const N = 5;
var transformed = channel.Pipe(X, async value => /* transformation */, N);

// Using named arguments for clarity
transformed = channel.Pipe(
    maxConcurrency: X,
    capacity: N,
    transform: async value => /* transformation */);