Final tweaks before .NET 3

Author: Oren (electricessence)

Open.ChannelExtensions/Extensions.Pipe.cs

@@ -92,7 +92,7 @@ public static ChannelReader<TOut> PipeAsync<TWrite, TRead, TOut>(this Channel<TW
 		/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
 		/// <param name="cancellationToken">An optional cancellation token.</param>
 		/// <returns>The channel reader containing the output.</returns>
-		public static ChannelReader<TOut> PipeAsync<TIn, TOut>(this ChannelReader<TIn> source,
+		public static ChannelReader<TOut> TaskPipeAsync<TIn, TOut>(this ChannelReader<TIn> source,
 			int maxConcurrency, Func<TIn, Task<TOut>> transform, int capacity = -1, bool singleReader = false,
 			CancellationToken cancellationToken = default)
 			=> source.PipeAsync(maxConcurrency, e => new ValueTask<TOut>(transform(e)), capacity, singleReader, cancellationToken);
@@ -110,10 +110,10 @@ public static ChannelReader<TOut> PipeAsync<TIn, TOut>(this ChannelReader<TIn> s
 		/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
 		/// <param name="cancellationToken">An optional cancellation token.</param>
 		/// <returns>The channel reader containing the output.</returns>
-		public static ChannelReader<TOut> PipeAsync<TWrite, TRead, TOut>(this Channel<TWrite, TRead> source,
+		public static ChannelReader<TOut> TaskPipeAsync<TWrite, TRead, TOut>(this Channel<TWrite, TRead> source,
 			int maxConcurrency, Func<TRead, Task<TOut>> transform, int capacity = -1, bool singleReader = false,
 			CancellationToken cancellationToken = default)
-			=> source.Reader.PipeAsync(maxConcurrency, transform, capacity, singleReader, cancellationToken);
+			=> source.Reader.TaskPipeAsync(maxConcurrency, transform, capacity, singleReader, cancellationToken);
 
 		/// <summary>
 		/// Reads all entries and applies the values to the provided transform function before buffering the results into another channel for consumption.
@@ -194,7 +194,7 @@ public static ChannelReader<TOut> PipeAsync<TWrite, TRead, TOut>(this Channel<TW
 		/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
 		/// <param name="cancellationToken">An optional cancellation token.</param>
 		/// <returns>The channel reader containing the output.</returns>
-		public static ChannelReader<TOut> PipeAsync<TIn, TOut>(this ChannelReader<TIn> source,
+		public static ChannelReader<TOut> TaskPipeAsync<TIn, TOut>(this ChannelReader<TIn> source,
 			Func<TIn, Task<TOut>> transform, int capacity = -1, bool singleReader = false,
 			CancellationToken cancellationToken = default)
 			=> source.PipeAsync(e => new ValueTask<TOut>(transform(e)), capacity, singleReader, cancellationToken);
@@ -211,10 +211,10 @@ public static ChannelReader<TOut> PipeAsync<TIn, TOut>(this ChannelReader<TIn> s
 		/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
 		/// <param name="cancellationToken">An optional cancellation token.</param>
 		/// <returns>The channel reader containing the output.</returns>
-		public static ChannelReader<TOut> PipeAsync<TWrite, TRead, TOut>(this Channel<TWrite, TRead> source,
+		public static ChannelReader<TOut> TaskPipeAsync<TWrite, TRead, TOut>(this Channel<TWrite, TRead> source,
 			Func<TRead, Task<TOut>> transform, int capacity = -1, bool singleReader = false,
 			CancellationToken cancellationToken = default)
-			=> source.Reader.PipeAsync(transform, capacity, singleReader, cancellationToken);
+			=> source.Reader.TaskPipeAsync(transform, capacity, singleReader, cancellationToken);
 
 		/// <summary>
 		/// Reads all entries and applies the values to the provided transform function before buffering the results into another channel for consumption.

Open.ChannelExtensions/Extensions.ReadConcurrently.cs

@@ -40,16 +40,8 @@ public static Task<long> ReadAllConcurrentlyAsync<T>(this ChannelReader<T> reade
 			return Task
 				.WhenAll(readers)
 				.ContinueWith(
-					t =>
-					{
-						if (t.IsFaulted)
-							return Task.FromException<long>(t.Exception);
-						if (t.IsCanceled)
-							return Task.FromCanceled<long>(cancellationToken);
-						return Task.FromResult(t.Result.Sum());
-					},
-					TaskContinuationOptions.ExecuteSynchronously)
-				.Unwrap();
+					t => t.Result.Sum(),
+					TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously);
 
 			ValueTask ParallelReceiver(T item, long i) => receiver(item);
 		}

Open.ChannelExtensions/Extensions.WriteConcurrently.cs

@@ -47,17 +47,14 @@ public static Task<long> WriteAllConcurrentlyAsync<T>(this ChannelWriter<T> targ
 				.ContinueWith(t =>
 				{
 					if (complete)
-						target.Complete();
+						target.Complete(t.Exception);
 
-					if (t.IsFaulted)
-						return Task.FromException<long>(t.Exception);
-
-					if (t.IsCanceled)
-						return Task.FromCanceled<long>(cancellationToken);
-
-					return Task.FromResult(t.Result.Sum());
+					return t;
 				}, TaskContinuationOptions.ExecuteSynchronously)
-				.Unwrap();
+				.Unwrap()
+				.ContinueWith(
+					t => t.Result.Sum(),
+					TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously);
 
 			// returns false if there's no more (wasn't cancelled).
 			async Task<long> WriteAllAsyncCore()

Open.ChannelExtensions/Open.ChannelExtensions.csproj

@@ -1,25 +1,27 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFramework>netstandard2.0</TargetFramework>
+    <TargetFrameworks>netstandard2.0</TargetFrameworks>
     <LangVersion>latest</LangVersion>
     <Authors>electricessence</Authors>
-    <Description>A set of extensions for optimizing/simplifying System.Threading.Channels usage.
+    <Description>
+      A set of extensions for optimizing/simplifying System.Threading.Channels usage.
 
-Includes:
-ReadUntilCancelled, ReadAll, ReadAllConcurrently, WriteAll, WriteAllConcurrently, and Pipe operations.
+      Includes:
+      ReadUntilCancelled, ReadAll, ReadAllConcurrently, WriteAll, WriteAllConcurrently, and Pipe operations.
 
-Part of the "Open" set of libraries.</Description>
+      Part of the "Open" set of libraries.
+    </Description>
     <PackageProjectUrl>https://github.com/electricessence/Open.ChannelExtensions</PackageProjectUrl>
     <PackageLicenseUrl></PackageLicenseUrl>
     <Copyright>https://github.com/electricessence/Open.ChannelExtensions/blob/master/LICENSE</Copyright>
     <RepositoryType>git</RepositoryType>
     <PackageTags>channels dotnet threading tasks extensions</PackageTags>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-    <Version>2.4.1</Version>
+    <Version>2.5.0</Version>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
-    <AssemblyVersion>2.4.1.0</AssemblyVersion>
-    <FileVersion>2.4.1.0</FileVersion>
+    <AssemblyVersion>2.5.0.0</AssemblyVersion>
+    <FileVersion>2.5.0.0</FileVersion>
     <PackageReleaseNotes></PackageReleaseNotes>
   </PropertyGroup>