Enhanced async exception handling and tests

- Added two unit tests in `BasicTests.cs` for verifying exception handling in async operations. These tests focus on ensuring `OperationCanceledException` is properly thrown and handled in custom async receivers.
- Introduced a using directive for `System.Net.Mime.MediaTypeNames` in `Extensions.Pipe.cs` for consistency or new functionality support.
- Improved exception handling in the async pipeline within `Extensions.Pipe.cs`. The update specifically enhances the handling of `OperationCanceledException` in task continuations, ensuring exceptions are correctly identified and managed.
- Updated project metadata in `Open.ChannelExtensions.csproj`, incrementing the version to 8.4.2. This update includes changes to the package release notes, mentioning the addition of new `.Merge`, `.PipeAsync`, and `.PropagateCompletion` extensions, indicating preparation for a new library release.

Author: electricessence

Open.ChannelExtensions.Tests/BasicTests.cs

@@ -513,4 +513,37 @@ public static async Task PipeFilterAsyncTest()
 		even.Should().OnlyContain(e => e % 2 == 0);
 		odd.Should().OnlyContain(e => e % 2 != 0);
 	}
+
+	[Fact]
+	public static async Task ChannelReadAllAsyncExceptionTest()
+	{
+		static ValueTask<int> AsyncReceiver()
+			=> throw new OperationCanceledException("test exception");
+
+		// Should throw.
+		await Assert.ThrowsAsync<OperationCanceledException>(
+			async () => await Enumerable.Repeat(0, 10)
+				.ToChannel()
+				.ReadAllAsync(async _ => await AsyncReceiver())
+		);
+	}
+
+	[Fact]
+	public static async Task PipeAsyncReadAllExceptionTest()
+	{
+		static ValueTask<int> AsyncReceiver()
+			=> throw new OperationCanceledException("test exception");
+
+		// Should also throw.
+		await Assert.ThrowsAsync<OperationCanceledException>(
+			async () =>
+			{
+				var reader = Enumerable.Repeat(0, 10)
+					.ToChannel()
+					.PipeAsync(1, async _ => await AsyncReceiver());
+
+				_ = await reader.ReadAll(_ => { });
+			});
+	}
+
 }

Open.ChannelExtensions/Extensions.Pipe.cs

@@ -1,4 +1,6 @@
-namespace Open.ChannelExtensions;
+using static System.Net.Mime.MediaTypeNames;
+
+namespace Open.ChannelExtensions;
 
 [SuppressMessage("Roslynator", "RCS1047:Non-asynchronous method name should not end with 'Async'.", Justification = "In order to differentiate between non async versions.")]
 public static partial class Extensions
@@ -98,7 +100,22 @@ public static ChannelReader<TOut> PipeAsync<TIn, TOut>(this ChannelReader<TIn> s
 					: ValueNotReady(result, cancellationToken); // Result is not ready, so we need to wait for it.
 			}, cancellationToken)
 			.ContinueWith(
-				t => writer.Complete(t.Exception),
+				t =>
+				{
+					Exception? ex = t.Exception;
+
+					/* We could have been cancelled
+					 * via a thrown cancel exception
+					 * that is not triggered by the token. */
+					if (ex is null
+						&& t.Status is TaskStatus.Canceled
+						&& !cancellationToken.IsCancellationRequested)
+					{
+						ex = new OperationCanceledException();
+					}
+
+					writer.Complete(ex);
+				},
 				CancellationToken.None,
 				TaskContinuationOptions.ExecuteSynchronously,
 				TaskScheduler.Current);

Open.ChannelExtensions/Open.ChannelExtensions.csproj

@@ -22,7 +22,7 @@
 		<RepositoryType>git</RepositoryType>
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
 		<GenerateDocumentationFile>true</GenerateDocumentationFile>
-		<Version>8.4.1</Version>
+		<Version>8.4.2</Version>
 		<PackageReleaseNotes>Added .Merge, .PipeAsync, and .PropagateCompletion extensions.</PackageReleaseNotes>
 		<PackageLicenseExpression>MIT</PackageLicenseExpression>
 		<PublishRepositoryUrl>true</PublishRepositoryUrl>