Open.ChannelExtensions 3.5.0

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.

Install-Package Open.ChannelExtensions -Version 3.5.0
dotnet add package Open.ChannelExtensions --version 3.5.0
<PackageReference Include="Open.ChannelExtensions" Version="3.5.0" />
For projects that support PackageReference, copy this XML node into the project file to reference the package.
paket add Open.ChannelExtensions --version 3.5.0
The NuGet Team does not provide support for this client. Please contact its maintainers for support.
#r "nuget: Open.ChannelExtensions, 3.5.0"
#r directive can be used in F# Interactive, C# scripting and .NET Interactive. Copy this into the interactive tool or source code of the script to reference the package.
// Install Open.ChannelExtensions as a Cake Addin
#addin nuget:?package=Open.ChannelExtensions&version=3.5.0

// Install Open.ChannelExtensions as a Cake Tool
#tool nuget:?package=Open.ChannelExtensions&version=3.5.0
The NuGet Team does not provide support for this client. Please contact its maintainers for support.

Open.ChannelExtensions

NuGet

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

  • Filter
  • Transform
  • Batch
  • Join

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*/)
    .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 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 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 X = 4;
const N = 5;
var transformed = channel.Pipe(
    X, async value => /* transformation */, N);

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

Open.ChannelExtensions

NuGet

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

  • Filter
  • Transform
  • Batch
  • Join

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*/)
    .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 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 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 X = 4;
const N = 5;
var transformed = channel.Pipe(
    X, async value => /* transformation */, N);

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

NuGet packages (3)

Showing the top 3 NuGet packages that depend on Open.ChannelExtensions:

Package Downloads
Open.Database.Extensions.Channel
Database extensions for pipelining data through channels. Includes Open.Database.Extensions.Core.
DataPipe.Core
Provides AWS support to the DataPipe platform. Reference this package if you need to read or write to S3 from a data pipe script
Pype.Background.Channels
Package Description

GitHub repositories (1)

Showing the top 1 popular GitHub repositories that depend on Open.ChannelExtensions:

Repository Stars
WolvenKit/WolvenKit
Mod editor/creator for RED Engine games. The point is to have an all in one tool for creating mods for the games made with the engine.

Version History

Version Downloads Last updated
3.5.0 16,783 7/16/2020
3.4.0 2,553 6/3/2020
3.3.2 203 6/1/2020
3.3.1 220 5/14/2020
3.3.0 2,482 2/19/2020
2.6.0 206 10/20/2019