Open.ChannelExtensions: pipelines over System.Threading.Channels

Extension methods that turn a channel into a pipeline

Open.ChannelExtensions is a set of extensions for optimizing and simplifying System.Threading.Channels usage, and the whole thing is a NuGet package with a detailed API documentation site linked from the readme.

The starting point is that System.Threading.Channels gives you a low level primitive. You create a channel, you write to it, you read from it, and you handle the backpressure and the completion yourself. That is the right abstraction for a producer and consumer pair, but it means a pipeline of several stages is a lot of explicit plumbing.

This package adds the pipeline vocabulary on top. The read and write group comes first, with optional concurrency levels. It can read all entries in a channel, write all entries from a source to a channel, and pipe, meaning consume, all entries to a buffer. There is also support for AsAsyncEnumerable, giving an IAsyncEnumerable, and that works for .NET Standard 2.1 and later and .NET Core 3 and later.

The second group is the set of special ChannelReader operations, and these are the ones that do not exist in the base library. Filter reads from the channel until a match is found. Transform applies a transform function upon successfully reading an item from the channel. Batch attempts to group items into a List or a Queue before they become available for reading. And Join combines batches into a single channel.

The pipeline example is what shows how these compose, and it reads as a single expression rather than as a sequence of statements:

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.
    });

So a bounded channel of ten is created, filled from an async source, piped through one transform with a concurrency of two and a capacity of five, then through a second transform with a capacity of three, and finally read out. The whole thing is one awaitable expression, and the parameters are named where the meaning would otherwise be unclear.

There is a second form of the same thing that starts from a plain sequence rather than a channel, and the readme shows it as a variant where the source is an IEnumerable converted with ToChannel, with the bounded size and a single reader flag. The order of the transforms is reversed between the two forms, which is a small detail that matters when reading the examples.

The project is MIT licensed, the primary language is C#, and the default branch is master. The releases are v6.3 in January 2024, described as the final V6 release, v7.0 in the same month, and v9.0 in November 2024 described as a .NET 9 refresh. The last push was on 2026-07-10.

Reading and writing, and the four ways to drain a channel

The reading section covers the range of ways to consume a channel, and the variations are all about two independent choices: whether the processing is synchronous or asynchronous, and whether you get the entry index.

The simplest form reads one by one until the channel is closed and calls the callback with the entry:

await channel.ReadAllAsync(
    async entry => { await /* Processing Code */ });

There is a variant that passes the entry and the index, and an async variant that awaits the callback, and an async variant that awaits the callback and receives the index as well. So four combinations from two switches.

Then there is the concurrent form, ReadAllConcurrently, which takes a maximum concurrency and processes entries in parallel rather than one at a time. The async counterpart is ReadAllConcurrentlyAsync.

On the writing side the shape is similar. WriteAll dumps a source enumeration into the channel, and the complete flag decides whether the channel is closed when the source is empty. The async version takes a source that is an IEnumerable of Task or of ValueTask, which is the case where producing an item is itself asynchronous:

// source can be any IEnumerable<Task<T>> or IEnumerable<ValueTask<T>>.
await channel.WriteAllAsync(source, complete: true);

And WriteAllConcurrentlyAsync synchronizes reading from the source and processes the results concurrently, taking the maximum concurrency, the source and the complete flag.

So the writing side has the same two axes: sync or async source, and serial or concurrent consumption. The complete flag is the one that requires care, because forgetting it leaves the channel open and a reader waiting forever, so it is explicit in every signature rather than defaulted silently.

The naming convention is worth noting because it is consistent. The Async suffix means the delegate you pass is awaited, and the Concurrently suffix means the work happens in parallel. The two combine, which is why there are both ReadAllConcurrentlyAsync and WriteAllConcurrentlyAsync.

Filter and Transform, and the warning about swallowed errors

The Filter and Transform extensions are described as operating synchronously after an item is read from the channel, and they chain off the reader rather than off the channel:

// Filter and transform when reading.
channel.Reader
    .Filter(predicate) // .Where()
    .Transform(selector) // .Select()
    .ReadAllAsync(async value => {/*...*/});

So the shape is channel.Reader.Filter then Transform then a read, and the comments in the example name the LINQ equivalents, with Filter commented as the equivalent of Where and Transform as the equivalent of Select.

The warning attached to them is the most important line in that section, and it is worth reading exactly: any predicate or selector function must trap errors, of the downstream read will fail and data may not be recoverable.

That is a genuine hazard rather than boilerplate. Both operations run inline in the read path, so an exception thrown by the predicate propagates out of the read itself. Once that happens the channel is not in a clean state, and the entries that were in flight may be lost, which is why the advice is to catch inside the function rather than outside it.

This is also the clearest illustration of the design philosophy of the package. The extensions are thin and they put the work in the caller's delegate, which means they stay fast and they stay out of the way, and the caller carries the responsibility for error handling in the delegate.

The same shape appears in the batching and joining sections. Batch groups into a List and comes with a WithTimeout option, and the example comment explains that any non-empty batches are flushed every second, so a batch is not held indefinitely waiting to fill:

values.Reader
    .Batch(10 /*batch size*/) // Groups into List<T>.
    .WithTimeout(1000) // Any non-empty batches are flushed every second.
    .ReadAllAsync(async batch => {/*...*/});

And Join is described as the inverse of batching, combining the batches into a single channel. So the two operations together give you a windowing pattern, and the timeout is what stops the window from stalling when the source goes quiet.

The Pipe overloads, and how the parameters multiply

The pipelining section shows four overloads of Pipe, and the pattern is that each one adds an option and the count grows.

The simplest transforms values in a source channel into a new unbounded channel. The next adds a maximum concurrency of X and takes the concurrency as the first argument. The third transforms into a new bounded channel with a bound of N entries, taking the bound as the last argument. And the fourth combines both, taking the concurrency and then the bound.

There is also a named-argument form of the last one, which is the version to reach for when the arguments are not obvious, with maxConcurrency, capacity and transform named explicitly.

The result of all this is that the signature of a pipeline stage has four possible shapes, and the compiler picks between them from the argument types. It is a readable pattern in a language where overload resolution is well understood, and the readme shows the named form precisely because that is the one you can read without counting parameters.

The unbounded versus bounded distinction is the substantive one. An unbounded channel never blocks the producer, which is right when the downstream is fast or when losing entries is acceptable. A bounded channel applies backpressure, which is right when you need the producer to slow down rather than allocate without limit. The concurrency argument is a third axis, and it controls how many items are in flight through the transform at once.

So each stage in a pipeline has three independent dials, and the package exposes all of them on the same method. That is the source of both its convenience and its complexity, and the reason the named-argument form exists.

The readme shows the transform functions as named parameters in the pipeline example, and the transform delegates are described as async in the concurrent case, which is consistent with the rest of the API where the async suffix marks an awaited delegate.

A benchmark project, and what the repository is actually for

The repository layout is more than a single library, and the directories are the clearest description of the project's scope.

There is the main Open.ChannelExtensions/ directory holding the library, and then three sibling projects with revealing names. Open.ChannelExtensions.Benchmarks/ is a benchmark project, Open.ChannelExtensions.Tests/ is the test project, and Open.ChannelExtensions.ComparisonTests/ is a separate project whose name implies it compares the extensions against something else, most likely against the equivalent hand-written channel code, since the point of an extension library is that the alternative is the same logic written out by hand.

That last directory is the interesting one for anyone evaluating the package, because a comparison test suite is a claim about the extension of the base library rather than a claim about the extension in isolation. If the extension is as fast as the hand-written version, then adopting it costs nothing in throughput and saves the plumbing.

There is a solution file, Open.ChannelExtensions.sln, at the root, which ties the four projects together. There is a docfx/ directory and a docs/ directory, which is how the API documentation site is produced, with the docfx configuration and the documentation sources kept apart.

The .editorconfig and .gitattributes files are present, and there is a .github/ directory for the workflows, plus the LICENSE and the README.

On maintenance, the version history is worth reading carefully because it is uneven. v6.3 in January 2024 is labelled the final V6 release, and v7.0 came out the same day, and then there is a gap until v9.0 in November 2024, labelled a .NET 9 refresh. So there is no v8 in the list, and the numbering jumps from 7 to 9.

The last push was on 2026-07-10, well after the newest tag. So the working tree is ahead of the last release, which for a library means the fixes and the additions since the .NET 9 refresh are not yet published to NuGet.

The library targets the .NET channel API, which arrived in .NET Core 3, and the readme notes the AsAsyncEnumerable support applies to .NET Standard 2.1 and later and .NET Core 3 and later. The topic tags are async, channel, channels, dotnet, extensions, tasks and threading, which is exactly the surface area of the package.

Editorial conclusion

Open.ChannelExtensions suits a .NET team with a multi-stage producer and consumer pipeline who would otherwise hand-write the channel plumbing for each stage, since the read, write, filter, transform and batch operations cover the shapes that come up repeatedly and the benchmark and comparison test projects in the repository suggest the performance cost of adopting them was measured. It does not suit you if you want to understand the underlying channel semantics from the library, because these are extensions over the base API rather than a replacement for it. Before you adopt it, note that the last published release is the v9.0 .NET 9 refresh while the repository was pushed to in July 2026, so check which fixes are in the package you install, and wrap any predicate or selector in its own error handling, because the readme says a throwing delegate fails the downstream read and may lose data.

Frequently asked questions

Official sources

1. License: MIT
2. Open-NET-Libraries/Open.ChannelExtensions on GitHub
3. Project website
4. README
5. Releases