A set of extensions for optimizing/simplifying System.Threading.Channels usage.
Includes:
ReadUntilCancelled, ReadAll, ReadAllConcurrently, WriteAll, WriteAllConcurrently, and Pipe operations.
Part of the "Open" set of libraries.
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
Both of these extensions operate synchronously after an item is read from the channel.
Any predicate or selector function must trap errors of the downstream read will fail and data may not be recoverable.
// Filter and transform when reading.
channel.Reader
.Filter(predicate) // .Where()
.Transform(selector) // .Select()
.ReadAllAsync(async value => {/*...*/});
Batching
values.Reader
.Batch(10 /*batch size*/) // Groups into List<T>.
.WithTimeout(1000) // Any non-empty batches are flushed every second.
.ReadAllAsync(async batch => {/*...*/});
Joining
The inverse of batching.
batches.Reader
.Join() // Combines the batches into a single channel.
.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 */);
