Open.ChannelExtensions
A set of extensions for optimizing/simplifying System.Threading.Channels usage.
Click here for detailed documentation.
Highlights
Read & Write
With optional concurrency levels.
- Reading all entries in a channel.
- Writing all entries from a source to a channel.
- Piping (consuming) all entries to a buffer (channel).
.AsAsyncEnumerable()(IAsyncEnumerable) support for .NET Standard 2.1+ and .NET Core 3+
Special ChannelReader Operations
FilterTransformBatchJoin
Installation
Install-Package Open.ChannelExtensions
Examples
Being able to define an asynchronous pipeline with best practice usage using simple expressive syntax:
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.
});await source /* IEnumerable<T> */
.ToChannel(boundedSize: 10, singleReader: true)
.PipeAsync(asyncTransform01, /* capacity */ 5)
.Pipe(
maxConcurrency: 2,
capacity: 3,
transform: transform02)
.ReadAll(finalTransformedValue => {
// Do something with each final value.
});Reading (until the channel is closed)
One by one read each entry from the channel
await channel.ReadAll(
entry => { /* Processing Code */ });await channel.ReadAll(
(entry, index) => { /* Processing Code */ });await channel.ReadAllAsync(
async entry => { await /* Processing Code */ });await channel.ReadAllAsync(
async (entry, index) => { await /* Processing Code */ });Read concurrently each entry from the channel
await channel.ReadAllConcurrently(
maxConcurrency,
entry => { /* Processing Code */ });await channel.ReadAllConcurrentlyAsync(
maxConcurrency,
async entry => { await /* Processing Code */ });Writing
If complete is true, the channel will be closed when the source is empty.
Dump a source enumeration into the channel
// source can be any IEnumerable<T>.
await channel.WriteAll(source, complete: true);// source can be any IEnumerable<Task<T>> or IEnumerable<ValueTask<T>>.
await channel.WriteAllAsync(source, complete: true);Synchronize reading from the source and process the results concurrently
// source can be any IEnumerable<Task<T>> or IEnumerable<ValueTask<T>>.
await channel.WriteAllConcurrentlyAsync(
maxConcurrency, source, complete: true);Filter & Transform
// Filter and transform when reading.
channel.Reader
.Filter(predicate) // .Where()
.Transform(selector) // .Select()
.ReadAllAsync(async value => {/*...*/});Batching
values.Reader
.Batch(10 /*batch size*/)
.WithTimeout(1000) // Any non-empty batches are flushed every second.
.ReadAllAsync(async batch => {/*...*/});Joining
batches.Reader
.Join()
.ReadAllAsync(async value => {/*...*/});Pipelining / Transforming
Transform and buffer entries
// Transform values in a source channel to new unbounded channel.
var transformed = channel.Pipe(
async value => /* transformation */);// Transform values in a source channel to new unbounded channel with a max concurrency of X.
const int X = 4;
var transformed = channel.Pipe(
X, async value => /* transformation */);// Transform values in a source channel to new bounded channel bound of N entries.
const int N = 5;
var transformed = channel.Pipe(
async value => /* transformation */, N);// Transform values in a source channel to new bounded channel bound of N entries with a max concurrency of X.
const int X = 4;
const int N = 5;
var transformed = channel.Pipe(
X, async value => /* transformation */, N);
// or
transformed = channel.Pipe(
maxConcurrency: X,
capacity: N,
transform: async value => /* transformation */);