Enhanced async cancellation handling

- Updated the expected exception type in `CancellationTests` from `OperationCanceledException` to `TaskCanceledException`, reflecting a more accurate expectation for task cancellation scenarios.
- Added a new extension method `WaitToReadOrCancelAsync` in `Extensions.Read.cs` to improve support for asynchronous read operations with cancellation.
- Refactored existing reading logic to utilize the new `WaitToReadOrCancelAsync` method, streamlining cancellation handling.
- Incremented the version of `Open.ChannelExtensions` to 8.4.0, indicating the introduction of new features and improvements.

Author: electricessence

Open.ChannelExtensions.Tests/CancellationTests.cs

@@ -57,7 +57,7 @@ await range
 		}
 		catch (Exception ex)
 		{
-			Assert.IsType<OperationCanceledException>(ex);
+			Assert.IsType<TaskCanceledException>(ex);
 		}
 
 		Assert.Equal(1, count);

Open.ChannelExtensions/Extensions.Read.cs

@@ -6,6 +6,24 @@ public static partial class Extensions
 {
 	private const string MustBeAtLeast1 = "Must be at least 1.";
 
+	/// <summary>
+	/// Waits for read to complete or the cancellation token to be cancelled.
+	/// </summary>
+	public static async ValueTask<bool> WaitToReadOrCancelAsync<T>(this ChannelReader<T> reader, CancellationToken cancellationToken)
+	{
+		if (reader is null) throw new ArgumentNullException(nameof(reader));
+		Contract.EndContractBlock();
+
+		try
+		{
+			return await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false);
+		}
+		catch (OperationCanceledException)
+		{
+			return false;
+		}
+	}
+
 	/// <summary>
 	/// Creates an enumerable that will read from the channel until no more are available for read.
 	/// </summary>
@@ -117,38 +135,32 @@ public static async ValueTask<long> ReadUntilCancelledAsync<T>(this ChannelReade
 			await Task.Yield();
 
 		long index = 0;
-		try
+
+		if (cancellationToken.CanBeCanceled)
 		{
-			if (cancellationToken.CanBeCanceled)
+			do
 			{
-				do
-				{
-					while (
-						!cancellationToken.IsCancellationRequested
-						&& reader.TryRead(out T? item))
-					{
-						await receiver(item, index++).ConfigureAwait(false);
-					}
-				}
 				while (
 					!cancellationToken.IsCancellationRequested
-					&& await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false));
-			}
-			else
-			{
-				do
+					&& reader.TryRead(out T? item))
 				{
-					while (reader.TryRead(out T? item))
-					{
-						await receiver(item, index++).ConfigureAwait(false);
-					}
+					await receiver(item, index++).ConfigureAwait(false);
 				}
-				while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false));
 			}
+			while (
+				!cancellationToken.IsCancellationRequested
+				&& await reader.WaitToReadOrCancelAsync(cancellationToken).ConfigureAwait(false));
 		}
-		catch (OperationCanceledException)
+		else
 		{
-			// In case WaitToReadAsync is cancelled.
+			do
+			{
+				while (reader.TryRead(out T? item))
+				{
+					await receiver(item, index++).ConfigureAwait(false);
+				}
+			}
+			while (await reader.WaitToReadOrCancelAsync(cancellationToken).ConfigureAwait(false));
 		}
 
 		return index;
@@ -373,27 +385,20 @@ public static async ValueTask ReadAllAsEnumerablesAsync<T>(this ChannelReader<T>
 		if (deferredExecution)
 			await Task.Yield();
 
-		try
+		if (cancellationToken.CanBeCanceled)
 		{
-			if (cancellationToken.CanBeCanceled)
-			{
-				while (
-					!cancellationToken.IsCancellationRequested
-					&& await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
-				{
-					await receiver(reader.ReadAvailable(cancellationToken)).ConfigureAwait(false);
-				}
-				return;
-			}
-
-			while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
+			while (
+				!cancellationToken.IsCancellationRequested
+				&& await reader.WaitToReadOrCancelAsync(cancellationToken).ConfigureAwait(false))
 			{
 				await receiver(reader.ReadAvailable(cancellationToken)).ConfigureAwait(false);
 			}
+			return;
 		}
-		catch (OperationCanceledException)
+
+		while (await reader.WaitToReadOrCancelAsync(cancellationToken).ConfigureAwait(false))
 		{
-			// In case WaitToReadAsync is cancelled.
+			await receiver(reader.ReadAvailable(cancellationToken)).ConfigureAwait(false);
 		}
 	}
 

Open.ChannelExtensions/Open.ChannelExtensions.csproj

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