Fixed issue with inconsistent exception handling when batching.

Author: electricessence

Open.ChannelExtensions.Tests/PipelineExceptionTests.cs

@@ -0,0 +1,73 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using System.Threading.Channels;
+using Xunit;
+
+namespace Open.ChannelExtensions.Tests;
+public static class PipelineExceptionTests
+{
+	const int BatchSize = 20;
+	const int Elements = 100;
+
+	[Theory]
+	[InlineData(BatchSize - 1)]
+	[InlineData(BatchSize)]
+	[InlineData(BatchSize + 1)]
+	[InlineData(-1)]
+	[InlineData(Elements)]
+	public static async Task Regular(int elementToThrow)
+	{
+		var channel = Channel.CreateBounded<int>(10000);
+
+		for (int i = 0; i < Elements; i++)
+		{
+			await channel.Writer.WriteAsync(i);
+		}
+
+		var task = channel.Reader
+			.Pipe(1, element => elementToThrow == -1 || element == elementToThrow ? throw new Exception() : element)
+			.Pipe(2, evt => evt * 2)
+			//.Batch(20)
+			.PipeAsync(1, evt => new ValueTask<int>(evt))
+			.ReadAll(_ => { });
+
+		if(elementToThrow == Elements)
+			channel.Writer.Complete(new Exception());
+		else
+			channel.Writer.Complete();
+
+		await Assert.ThrowsAsync<AggregateException>(async () => await task);
+	}
+
+	[Theory]
+	[InlineData(0)]
+	[InlineData(1)]
+	[InlineData(BatchSize - 1)]
+	[InlineData(BatchSize)]
+	[InlineData(BatchSize + 1)]
+	[InlineData(-1)]
+	public static async Task Batched(int elementToThrow)
+	{
+		var channel = Channel.CreateBounded<int>(10000);
+
+		for (int i = 0; i < Elements; i++)
+		{
+			await channel.Writer.WriteAsync(i);
+		}
+
+		var task = channel.Reader
+			.Pipe(1, element => elementToThrow == -1 || element == elementToThrow ? throw new Exception() : element)
+			.Pipe(2, evt => evt * 2)
+			.Batch(20)
+			.PipeAsync(1, evt => new ValueTask<List<int>>(evt))
+			.ReadAll(_ => { });
+
+		if(elementToThrow == Elements)
+			channel.Writer.Complete(new Exception());
+		else
+			channel.Writer.Complete();
+
+		await Assert.ThrowsAsync<AggregateException>(async () => await task);
+	}
+}

Open.ChannelExtensions/BufferingChannelReader.cs

@@ -99,13 +99,21 @@ public override bool TryRead(out TOut item)
 	/// <inheritdoc />
 	public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
 	{
-		if (Buffer?.Reader.Completion.IsCompleted != false)
-			return new ValueTask<bool>(false);
+		var buffer = Buffer;
+		if(buffer is null) return new ValueTask<bool>(false);
+
+		var completion = buffer.Reader.Completion;
+		if (completion.IsCompleted)
+		{
+			return completion.IsFaulted
+				? new ValueTask<bool>(Task.FromException<bool>(completion.Exception))
+				: new ValueTask<bool>(false);
+		}
 
 		if (cancellationToken.IsCancellationRequested)
 			return new ValueTask<bool>(Task.FromCanceled<bool>(cancellationToken));
 
-		ValueTask<bool> b = Buffer.Reader.WaitToReadAsync(cancellationToken);
+		ValueTask<bool> b = buffer.Reader.WaitToReadAsync(cancellationToken);
 		return b.IsCompleted ? b : WaitToReadAsyncCore(b, cancellationToken);
 	}
 

Open.ChannelExtensions/Open.ChannelExtensions.csproj

@@ -22,7 +22,7 @@
 		<RepositoryType>git</RepositoryType>
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
 		<GenerateDocumentationFile>true</GenerateDocumentationFile>
-		<Version>6.2.0</Version>
+		<Version>6.2.1</Version>
 		<PackageReleaseNotes></PackageReleaseNotes>
 		<PackageLicenseExpression>MIT</PackageLicenseExpression>
 		<PublishRepositoryUrl>true</PublishRepositoryUrl>