ChannelUtils/Extensions.cs
/// <summary>
/// We process the data in the channel and send it into new one
/// </summary>
/// <param name="data"></param>
/// <param name="action"></param>
/// <param name="exceptionalChannel"></param>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
We process the data in the channel and send it into new one
We process the data in the channel and send it into new one
public static ChannelReader<T> Process<T>(
			this ChannelReader<T> data,
			Action<T> action,
			ChannelWriter<DataProcessException>? exceptionalChannel = null)
		{
			var ch = Channel.CreateUnbounded<T>(new UnboundedChannelOptions
			{
				AllowSynchronousContinuations = true
			});
			Task.Run(async () =>
			{
				while (await data.WaitToReadAsync())
				{
					if (!data.TryRead(out var channelData)) continue;
					try
					{
						action(channelData);
						await ch.Writer.WriteAsync(channelData);
					}
					catch (DataProcessException processException) when (exceptionalChannel != null)
					{
						await exceptionalChannel.WriteAsync(processException);
					}
					catch (Exception exception) when (exceptionalChannel != null)
					{
						await exceptionalChannel.WriteAsync(new DataProcessException(exception));
					}
					catch
					{
					}
				}
				ch.Writer.Complete();
			});
			return ch.Reader;
		}
public static ChannelReader < T > Process < T > ( this ChannelReader < T > data , Action < T > action , ChannelWriter < DataProcessException > ? exceptionalChannel = null ) { var ch = Channel . CreateUnbounded < T > ( new UnboundedChannelOptions { AllowSynchronousContinuations = true } ) ; Task . Run ( async ( ) => { while ( await data . WaitToReadAsync ( ) ) { if ( ! data . TryRead ( out var channelData ) ) continue ; try { action ( channelData ) ; await ch . Writer . WriteAsync ( channelData ) ; } catch ( DataProcessException processException ) when ( exceptionalChannel != null ) { await exceptionalChannel . WriteAsync ( processException ) ; } catch ( Exception exception ) when ( exceptionalChannel != null ) { await exceptionalChannel . WriteAsync ( new DataProcessException ( exception ) ) ; } catch { } } ch . Writer . Complete ( ) ; } ) ; return ch . Reader ; }
We process the data in the channel and send it into new one
We process the data in the channel and send it into new one
// ignored
None
ChannelReader<T>
this
action
Action<T>
None
exceptionalChannel
ChannelWriter<DataProcessException>?
None
data
None
typeparam
None