ChannelUtils/Extensions.cs
/// <summary>
/// Accumulate data from the channel into chunks and send it to the new channel
/// </summary>
/// <param name="data"></param>
/// <param name="count"></param>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
Accumulate data from the channel into chunks and send it to the new channel
Accumulate data from the channel into chunks and send it to the new channel
public static ChannelReader<IReadOnlyCollection<T>> CreateChunks<T>(
			this ChannelReader<T> data,
			int count)
		{
			var ch = Channel.CreateUnbounded<IReadOnlyCollection<T>>();
			Task.Run(async () =>
			{
				var cts = new CancellationTokenSource();
				var tempCollection = new ConcurrentQueue<T>();
				var timeContainer = new TimeContainer
				{
					LastFlush = DateTime.Now
				};
#pragma warning disable 4014
				Task.Run(async () =>
#pragma warning restore 4014
					{
						while (!cts.Token.IsCancellationRequested)
						{
							await Task.Delay(TimeSpan.FromMinutes(1), cts.Token);
							if (DateTime.Now - timeContainer.LastFlush < TimeSpan.FromMinutes(1)
							    || tempCollection.IsEmpty)
							{
								continue;
							}
							await ch.Writer.WriteAsync(tempCollection.DequeueAll(), cts.Token);
						}
					},
					cts.Token);
				while (await data.WaitToReadAsync(cts.Token))
				{
					if (!data.TryRead(out var channelData)) continue;
					tempCollection.Enqueue(channelData);
					if (tempCollection.Count != count) continue;
					await ch.Writer.WriteAsync(tempCollection.DequeueAll(), cts.Token);
					timeContainer.LastFlush = DateTime.Now;
				}
				await ch.Writer.WriteAsync(tempCollection.DequeueAll(), cts.Token);
				ch.Writer.Complete();
				cts.Cancel();
			});
			return ch.Reader;
		}
public static ChannelReader < IReadOnlyCollection < T > > CreateChunks < T > ( this ChannelReader < T > data , int count ) { var ch = Channel . CreateUnbounded < IReadOnlyCollection < T > > ( ) ; Task . Run ( async ( ) => { var cts = new CancellationTokenSource ( ) ; var tempCollection = new ConcurrentQueue < T > ( ) ; var timeContainer = new TimeContainer { LastFlush = DateTime . Now } ; pragma warning disable 4014 Task . Run ( async ( ) => pragma warning restore 4014 { while ( ! cts . Token . IsCancellationRequested ) { await Task . Delay ( TimeSpan . FromMinutes ( 1 ) , cts . Token ) ; if ( DateTime . Now - timeContainer . LastFlush < TimeSpan . FromMinutes ( 1 ) || tempCollection . IsEmpty ) { continue ; } await ch . Writer . WriteAsync ( tempCollection . DequeueAll ( ) , cts . Token ) ; } } , cts . Token ) ; while ( await data . WaitToReadAsync ( cts . Token ) ) { if ( ! data . TryRead ( out var channelData ) ) continue ; tempCollection . Enqueue ( channelData ) ; if ( tempCollection . Count != count ) continue ; await ch . Writer . WriteAsync ( tempCollection . DequeueAll ( ) , cts . Token ) ; timeContainer . LastFlush = DateTime . Now ; } await ch . Writer . WriteAsync ( tempCollection . DequeueAll ( ) , cts . Token ) ; ch . Writer . Complete ( ) ; cts . Cancel ( ) ; } ) ; return ch . Reader ; }
Accumulate data from the channel into chunks and send it to the new channel
Accumulate data from the channel into chunks and send it to the new channel
// We do not want to endlessly wait for the newest data in the channel if the channel is not completed
None
ChannelReader<T>
this
count
int
None
data
None
typeparam
None