Optimize async operations with default

- Replaced `new ValueTask()` with `default` across async operations for better performance.
- In `BasicTests.cs`, updated `ReadAllAsync` method calls for efficiency.
- `SourceTests.cs` now uses `Assert.ThrowsAsync` for testing task cancellations.
- In `Extensions.Read.cs`, renamed `index` to `count` and simplified async read logic.
- Removed unnecessary cancellation checks in some async operations.
- Applied `ValueTask` optimization in write extensions for consistency.
- Updated project version to 8.4.1, indicating minor improvements and optimizations.

Author: electricessence

Open.ChannelExtensions.Tests/BasicTests.cs

@@ -187,7 +187,7 @@ public static async Task ReadAllAsync(int testSize)
 			.ReadAllAsync(i =>
 			{
 				result.Add(i);
-				return new ValueTask();
+				return default;
 			});
 		sw.Stop();
 
@@ -221,7 +221,7 @@ public static async Task PipeToBounded(int testSize)
 			.ReadAllAsync(i =>
 			{
 				result.Add(i);
-				return new ValueTask();
+				return default;
 			});
 		sw.Stop();
 
@@ -255,7 +255,7 @@ public static async Task PipeToUnbound(int testSize)
 			.ReadAllAsync(i =>
 			{
 				result.Add(i);
-				return new ValueTask();
+				return default;
 			});
 		sw.Stop();
 

Open.ChannelExtensions.Tests/SourceTests.cs

@@ -15,7 +15,7 @@ public static async Task ToChannelCancelledAfterwriteStarts()
 		catch (OperationCanceledException)
 		{ }
 
-		await reader.ReadAll(_ => { });
+		await Assert.ThrowsAsync<OperationCanceledException>(() => reader.ReadAll(_ => { }).AsTask());
 		await Assert.ThrowsAsync<TaskCanceledException>(() => reader.Completion);
 	}
 
@@ -26,8 +26,7 @@ public static async Task ToChannelCancelledBeforeWriteStarts()
 		cts.Cancel();
 		var reader = Enumerable.Range(0, 10_000).ToChannel(10, true, cts.Token);
 
-		var count = await reader.ReadAll(_ => { });
-		Assert.Equal(0, count);
+		await Assert.ThrowsAsync<TaskCanceledException>(() => reader.ReadAll(_ => { }).AsTask());
 		await Assert.ThrowsAsync<TaskCanceledException>(() => reader.Completion);
 	}
 }

Open.ChannelExtensions/Extensions.Read.cs

@@ -102,7 +102,6 @@ public static async ValueTask<List<T>> ReadBatchAsync<T>(this ChannelReader<T> r
 				if (results.Count == max)
 					return results;
 
-				cancellationToken.ThrowIfCancellationRequested();
 			}
 			while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false));
 		}
@@ -134,8 +133,9 @@ public static async ValueTask<long> ReadUntilCancelledAsync<T>(this ChannelReade
 		if (deferredExecution)
 			await Task.Yield();
 
-		long index = 0;
+		long count = 0;
 
+		// Note: if the channel has complete with an OperationCanceledException, this will throw when waiting to read.
 		if (cancellationToken.CanBeCanceled)
 		{
 			do
@@ -144,7 +144,7 @@ public static async ValueTask<long> ReadUntilCancelledAsync<T>(this ChannelReade
 					!cancellationToken.IsCancellationRequested
 					&& reader.TryRead(out T? item))
 				{
-					await receiver(item, index++).ConfigureAwait(false);
+					await receiver(item, count++).ConfigureAwait(false);
 				}
 			}
 			while (
@@ -157,13 +157,13 @@ public static async ValueTask<long> ReadUntilCancelledAsync<T>(this ChannelReade
 			{
 				while (reader.TryRead(out T? item))
 				{
-					await receiver(item, index++).ConfigureAwait(false);
+					await receiver(item, count++).ConfigureAwait(false);
 				}
 			}
-			while (await reader.WaitToReadOrCancelAsync(cancellationToken).ConfigureAwait(false));
+			while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false));
 		}
 
-		return index;
+		return count;
 	}
 
 	/// <summary>
@@ -250,7 +250,7 @@ public static ValueTask<long> ReadUntilCancelled<T>(this ChannelReader<T> reader
 			(e, i) =>
 			{
 				receiver(e, i);
-				return new ValueTask();
+				return default;
 			},
 			deferredExecution);
 
@@ -297,7 +297,7 @@ public static ValueTask<long> ReadUntilCancelled<T>(this ChannelReader<T> reader
 			(e, _) =>
 			{
 				receiver(e);
-				return new ValueTask();
+				return default;
 			},
 			deferredExecution);
 
@@ -356,7 +356,7 @@ public static ValueTask ReadAllAsEnumerables<T>(this ChannelReader<T> reader,
 			e =>
 			{
 				receiver(e);
-				return new ValueTask();
+				return default;
 			},
 			deferredExecution,
 			cancellationToken);
@@ -396,7 +396,7 @@ public static async ValueTask ReadAllAsEnumerablesAsync<T>(this ChannelReader<T>
 			return;
 		}
 
-		while (await reader.WaitToReadOrCancelAsync(cancellationToken).ConfigureAwait(false))
+		while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
 		{
 			await receiver(reader.ReadAvailable(cancellationToken)).ConfigureAwait(false);
 		}
@@ -775,7 +775,7 @@ public static ValueTask<long> ReadAll<T>(this ChannelReader<T> reader,
 			.ReadAllAsync((e, i) =>
 			{
 				receiver(e, i);
-				return new ValueTask();
+				return default;
 			},
 			deferredExecution,
 			cancellationToken);
@@ -894,7 +894,7 @@ public static ValueTask<long> ReadAll<T>(this ChannelReader<T> reader,
 			(e, _) =>
 			{
 				receiver(e);
-				return new ValueTask();
+				return default;
 			},
 			deferredExecution,
 			cancellationToken);

Open.ChannelExtensions/Extensions.ReadConcurrently.cs

@@ -177,7 +177,7 @@ public static Task<long> ReadAllConcurrently<T>(this ChannelReader<T> reader,
 			e =>
 			{
 				receiver(e);
-				return new ValueTask();
+				return default;
 			},
 			cancellationToken);
 
@@ -321,7 +321,7 @@ public static Task ReadAllConcurrentlyAsEnumerables<T>(this ChannelReader<T> rea
 			e =>
 			{
 				receiver(e);
-				return new ValueTask();
+				return default;
 			},
 			cancellationToken);
 	}

Open.ChannelExtensions/Extensions.Write.cs

@@ -35,7 +35,7 @@ await target
 				.ConfigureAwait(false);
 
 			long count = 0;
-			var next = new ValueTask();
+			ValueTask next = default;
 			foreach (ValueTask<T> e in source)
 			{
 				T? value = await e.ConfigureAwait(false);
@@ -340,7 +340,7 @@ await target
 				.ConfigureAwait(false);
 
 			long count = 0;
-			var next = new ValueTask();
+			ValueTask next = default;
 			await foreach (T? value in source)
 			{
 				await next.ConfigureAwait(false);

Open.ChannelExtensions/Extensions.WriteConcurrently.cs

@@ -78,7 +78,7 @@ async Task<long> WriteAllAsyncCore()
 			{
 				await shouldWait.ConfigureAwait(false);
 				long count = 0;
-				var next = new ValueTask();
+				ValueTask next = default;
 				bool potentiallyCancelled = true; // if it completed and actually returned false, no need to bubble the cancellation since it actually completed.
 				while (!errorToken.IsCancellationRequested
 					&& !cancellationToken.IsCancellationRequested
@@ -88,7 +88,7 @@ async Task<long> WriteAllAsyncCore()
 					await next.ConfigureAwait(false);
 					count++;
 					next = target.TryWrite(value) // do this to avoid unnecessary early cancel.
-						? new ValueTask()
+						? default
 						: target.WriteAsync(value, cancellationToken);
 				}
 				await next.ConfigureAwait(false);

Open.ChannelExtensions/Extensions._.cs

@@ -65,7 +65,7 @@ static async ValueTask ThrowChannelClosedExceptionIfFalse(ValueTask<bool> write,
 	internal static ValueTask CancelAsync(this CancellationTokenSource source)
 	{
 		source.Cancel();
-		return new ValueTask();
+		return default;
 	}
 #endif
 
@@ -174,7 +174,7 @@ public static ValueTask WaitToWriteAndThrowIfClosedAsync<T>(this ChannelWriter<T
 			return ThrowChannelClosedExceptionIfFalse(waitForWrite, ifClosedMessage);
 
 		if (waitForWrite.Result)
-			return new ValueTask();
+			return default;
 
 		if (string.IsNullOrWhiteSpace(ifClosedMessage)) throw new ChannelClosedException();
 		throw new ChannelClosedException(ifClosedMessage);

Open.ChannelExtensions/Open.ChannelExtensions.csproj

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