ChannelUtils/Extensions.cs
/// <summary>
/// The last action on the channel data
/// </summary>
/// <param name="data"></param>
/// <param name="asyncAction"></param>
/// <param name="exceptionalChannel"></param>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
The last action on the channel data
The last action on the channel data
public static Task Action<T>(
			this ChannelReader<T> data,
			Func<T, Task> asyncAction,
			ChannelWriter<DataProcessException>? exceptionalChannel = null)
		{
			return Task.Run(async () =>
			{
				while (await data.WaitToReadAsync())
				{
					if (!data.TryRead(out var channelData)) continue;
					try
					{
						await asyncAction(channelData);
					}
					catch (DataProcessException processException) when (exceptionalChannel != null)
					{
						await exceptionalChannel.WriteAsync(processException);
					}
					catch (Exception exception) when (exceptionalChannel != null)
					{
						await exceptionalChannel.WriteAsync(new DataProcessException(exception));
					}
					catch
					{
					}
				}
			});
		}
public static Task Action < T > ( this ChannelReader < T > data , Func < T , Task > asyncAction , ChannelWriter < DataProcessException > ? exceptionalChannel = null ) { return Task . Run ( async ( ) => { while ( await data . WaitToReadAsync ( ) ) { if ( ! data . TryRead ( out var channelData ) ) continue ; try { await asyncAction ( channelData ) ; } catch ( DataProcessException processException ) when ( exceptionalChannel != null ) { await exceptionalChannel . WriteAsync ( processException ) ; } catch ( Exception exception ) when ( exceptionalChannel != null ) { await exceptionalChannel . WriteAsync ( new DataProcessException ( exception ) ) ; } catch { } } } ) ; }
The last action on the channel data
The last action on the channel data
// ignored
None
ChannelReader<T>
this
asyncAction
Func<T, Task>
None
exceptionalChannel
ChannelWriter<DataProcessException>?
None
data
None
typeparam
None