ChannelUtils/Extensions.cs
/// <summary>
/// We created <see cref="n"/> new channels and send to it all data from
/// <see cref="inputChannel"/> by round robin strategy
/// </summary>
/// <param name="inputChannel"></param>
/// <param name="n"></param>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
We created  new channels and send to it all data from
by round robin strategy
We created new channels and send to it all data from by round robin strategy
public static ChannelReader<T>[] InParallel<T>(this ChannelReader<T> inputChannel, int n)
		{
			var outputs = new Channel<T>[n];
			for (int i = 0; i < n; i++)
				outputs[i] = Channel.CreateUnbounded<T>();
			Task.Run(async () =>
			{
				var index = 0;
				await foreach (var item in inputChannel.ReadAllAsync())
				{
					await outputs[index].Writer.WriteAsync(item);
					index = (index + 1) % n;
				}
				foreach (var channel in outputs)
					channel.Writer.Complete();
			});
			return outputs.Select(ch => ch.Reader).ToArray();
		}
public static ChannelReader < T > [ ] InParallel < T > ( this ChannelReader < T > inputChannel , int n ) { var outputs = new Channel < T > [ n ] ; for ( int i = 0 ; i < n ; i ++ ) outputs [ i ] = Channel . CreateUnbounded < T > ( ) ; Task . Run ( async ( ) => { var index = 0 ; await foreach ( var item in inputChannel . ReadAllAsync ( ) ) { await outputs [ index ] . Writer . WriteAsync ( item ) ; index = ( index + 1 ) % n ; } foreach ( var channel in outputs ) channel . Writer . Complete ( ) ; } ) ; return outputs . Select ( ch => ch . Reader ) . ToArray ( ) ; }
We created  new channels and send to it all data from
by round robin strategy
We created new channels and send to it all data from by round robin strategy
None
ChannelReader<T>
this
n
int
None
inputChannel
None
typeparam
None