Open.ChannelExtensions 8.0.2

dotnet add package Open.ChannelExtensions --version 8.0.2
NuGet\Install-Package Open.ChannelExtensions -Version 8.0.2
This command is intended to be used within the Package Manager Console in Visual Studio, as it uses the NuGet module's version of Install-Package.
<PackageReference Include="Open.ChannelExtensions" Version="8.0.2" />
For projects that support PackageReference, copy this XML node into the project file to reference the package.
paket add Open.ChannelExtensions --version 8.0.2
#r "nuget: Open.ChannelExtensions, 8.0.2"
#r directive can be used in F# Interactive and Polyglot Notebooks. 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=8.0.2

// Install Open.ChannelExtensions as a Cake Tool
#tool nuget:?package=Open.ChannelExtensions&version=8.0.2

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

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 */);
Product Compatible and additional computed target framework versions.
.NET net5.0 was computed.  net5.0-windows was computed.  net6.0 is compatible.  net6.0-android was computed.  net6.0-ios was computed.  net6.0-maccatalyst was computed.  net6.0-macos was computed.  net6.0-tvos was computed.  net6.0-windows was computed.  net7.0 was computed.  net7.0-android was computed.  net7.0-ios was computed.  net7.0-maccatalyst was computed.  net7.0-macos was computed.  net7.0-tvos was computed.  net7.0-windows was computed.  net8.0 is compatible.  net8.0-android was computed.  net8.0-browser was computed.  net8.0-ios was computed.  net8.0-maccatalyst was computed.  net8.0-macos was computed.  net8.0-tvos was computed.  net8.0-windows was computed. 
.NET Core netcoreapp2.0 was computed.  netcoreapp2.1 was computed.  netcoreapp2.2 was computed.  netcoreapp3.0 was computed.  netcoreapp3.1 was computed. 
.NET Standard netstandard2.0 is compatible.  netstandard2.1 is compatible. 
.NET Framework net461 was computed.  net462 was computed.  net463 was computed.  net47 was computed.  net471 was computed.  net472 was computed.  net48 was computed.  net481 was computed. 
MonoAndroid monoandroid was computed. 
MonoMac monomac was computed. 
MonoTouch monotouch was computed. 
Tizen tizen40 was computed.  tizen60 was computed. 
Xamarin.iOS xamarinios was computed. 
Xamarin.Mac xamarinmac was computed. 
Xamarin.TVOS xamarintvos was computed. 
Xamarin.WatchOS xamarinwatchos was computed. 
Compatible target framework(s)
Included target framework(s) (in package)
Learn more about Target Frameworks and .NET Standard.

NuGet packages (13)

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

Package Downloads
Indice.Common

Package Description

Indice.Features.Identity.SignInLogs

Package Description

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

Resistance.Postgres

.NET toolkit for PostgreSQL

GitHub repositories (2)

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

Repository Stars
oskardudycz/EventSourcing.NetCore
Examples and Tutorials of Event Sourcing in .NET
Excel-DNA/Samples
Various sample projects and snippets related to Excel-DNA
Version Downloads Last updated
8.0.2 13,320 1/31/2024
8.0.1 925 1/26/2024
8.0.0 3,687 1/23/2024
7.0.2 1,631 1/31/2024
7.0.1 234 1/26/2024
7.0.0 332 1/23/2024
6.3.2 2,549 1/31/2024
6.3.1 1,027 1/26/2024
6.3.0 288 1/23/2024
6.2.2 272,285 10/5/2022
6.2.1 36,694 9/3/2022
6.2.0 37,892 7/13/2022
6.1.0 5,427 6/30/2022
6.0.3 38,028 5/12/2022
6.0.2 27,591 4/13/2022
6.0.1 17,901 3/1/2022
6.0.0 3,838 2/22/2022
5.2.1 9,260 2/22/2022
5.2.0 4,364 2/6/2022
5.1.3 48,089 10/12/2021
5.1.2 1,060 10/8/2021
5.1.1 720 10/8/2021
5.1.0 798 10/8/2021
5.0.0 36,659 6/16/2021
3.5.0 177,826 7/16/2020
3.4.0 5,459 6/3/2020
3.3.2 934 6/1/2020
3.3.1 972 5/14/2020
3.3.0 4,601 2/19/2020
2.6.0 933 10/20/2019