Added PropagateCompletion helper extensions.

Author: electricessence

Open.ChannelExtensions.Tests/PropagationTests.cs

@@ -0,0 +1,14 @@
+namespace Open.ChannelExtensions.Tests;
+public static class PropagationTests
+{
+	[Fact]
+	public static async Task PropagateCompleteionTest()
+	{
+		var inputChannel = Channel.CreateBounded<int>(100);
+		var outputChannel = Channel.CreateBounded<int>(100);
+
+		inputChannel.PropagateCompletion(outputChannel);
+		inputChannel.Writer.Complete();
+		await outputChannel.Reader.Completion;
+	}
+}

Open.ChannelExtensions/Extensions.Pipe.cs

@@ -34,15 +34,15 @@ public static async ValueTask<long> PipeTo<T>(this ChannelReader<T> source,
 		{
 			if (complete)
 			{
-				target.Complete(ex);
+				target.TryComplete(ex);
 				complete = false;
 			}
 			throw;
 		}
 		finally
 		{
 			if (complete)
-				target.Complete();
+				target.TryComplete();
 		}
 	}
 

Open.ChannelExtensions/Extensions._.cs

@@ -6,6 +6,9 @@ namespace Open.ChannelExtensions;
 /// <summary>
 /// Extensions for operating with System.Threading.Channels.
 /// </summary>
+[SuppressMessage("RCS1047FadeOut",
+	"RCS1047FadeOut: Async Naming",
+	Justification = "Some methods use the Async suffix to distinguish between the contents of their parameters.")]
 public static partial class Extensions
 {
 	internal static Channel<T> CreateChannel<T>(ChannelOptions channelOptions)
@@ -69,6 +72,52 @@ internal static IEnumerable<ValueTask<T>> WrapValueTask<T>(this IEnumerable<Func
 			yield return new ValueTask<T>(e());
 	}
 
+	/// <summary>
+	/// Will propagate the completion of the <paramref name="source"/> channel reader to the <paramref name="target"/> channel writer.
+	/// </summary>
+	/// <remarks>If the <paramref name="cancellationToken"/> is cancelled, the propagation will not occur.</remarks>
+	/// <returns>The source reader.</returns>
+	public static ChannelReader<TSource> PropagateCompletion<TSource, TTarget>(
+		this ChannelReader<TSource> source, ChannelWriter<TTarget> target, CancellationToken cancellationToken = default)
+	{
+		if (source is null) throw new ArgumentNullException(nameof(source));
+		if (target is null) throw new ArgumentNullException(nameof(target));
+		Contract.EndContractBlock();
+
+		if (cancellationToken.IsCancellationRequested)
+			return source;
+
+		source.Completion.ContinueWith(t => {
+			if (t.IsFaulted)
+				target.TryComplete(t.Exception);
+			else
+				target.TryComplete();
+		}, cancellationToken);
+
+		return source;
+	}
+
+	/// <inheritdoc cref="PropagateCompletion{TSource, TTarget}(ChannelReader{TSource}, ChannelWriter{TTarget}, CancellationToken)"/>
+	[ExcludeFromCodeCoverage]
+	public static ChannelReader<TSource> PropagateCompletion<TSource, TTargetIn, TTargetOut>(
+		this ChannelReader<TSource> source, Channel<TTargetIn, TTargetOut> target, CancellationToken cancellationToken = default)
+		=> PropagateCompletion(source, target.Writer, cancellationToken);
+
+	/// <inheritdoc cref="PropagateCompletion{TSource, TTarget}(ChannelReader{TSource}, ChannelWriter{TTarget}, CancellationToken)"/>
+	[ExcludeFromCodeCoverage]
+	public static Channel<TSourceIn, TSourceOut> PropagateCompletion<TSourceIn, TSourceOut, TTarget>(
+		this Channel<TSourceIn, TSourceOut> source, ChannelWriter<TTarget> target, CancellationToken cancellationToken = default)
+	{
+		_ = source.Reader.PropagateCompletion(target, cancellationToken);
+		return source;
+	}
+
+	/// <inheritdoc cref="PropagateCompletion{TSource, TTarget}(ChannelReader{TSource}, ChannelWriter{TTarget}, CancellationToken)"/>/>
+	[ExcludeFromCodeCoverage]
+	public static Channel<TSourceIn, TSourceOut> PropagateCompletion<TSourceIn, TSourceOut, TTargetIn, TTargetOut>(
+		this Channel<TSourceIn, TSourceOut> source, Channel<TTargetIn, TTargetOut> target, CancellationToken cancellationToken = default)
+		=> PropagateCompletion(source, target.Writer, cancellationToken);
+
 	/// <summary>
 	/// Uses <see cref="ChannelWriter{T}.WaitToWriteAsync(CancellationToken)"/> to peek and see if the channel can still be written to.
 	/// </summary>
@@ -271,13 +320,7 @@ public static ChannelReader<T> ToChannel<T>(this IEnumerable<T> source,
 	/// <summary>
 	/// Writes all entries from the source to a channel and calls complete when finished.
 	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
-	/// <param name="source">The source data to use.</param>
-	/// <param name="capacity">The optional bounded capacity of the channel. Default is unbound.</param>
-	/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
-	/// <param name="deferredExecution">If true, calls await Task.Yield() before writing to the channel.</param>
-	/// <param name="cancellationToken">An optional cancellation token.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, int, bool, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannel<T>(this IAsyncEnumerable<T> source,
 		int capacity = -1, bool singleReader = false,
 		bool deferredExecution = false,
@@ -288,29 +331,20 @@ public static ChannelReader<T> ToChannel<T>(this IAsyncEnumerable<T> source,
 	/// <summary>
 	/// Writes all entries from the source to a channel and calls complete when finished.
 	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
-	/// <param name="source">The source data to use.</param>
-	/// <param name="channelOptions">The options for configuring the new channel.</param>
-	/// <param name="deferredExecution">If true, calls await Task.Yield() before writing to the channel.</param>
-	/// <param name="cancellationToken">An optional cancellation token.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, ChannelOptions, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannel<T>(this IAsyncEnumerable<T> source,
 		ChannelOptions channelOptions,
 		bool deferredExecution = false,
 		CancellationToken cancellationToken = default)
 		=> CreateChannel<T>(channelOptions)
 			.Source(source, deferredExecution, cancellationToken);
 
-	/// <summary>
-	/// Writes all entries from the source to a channel and calls complete when finished.
-	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
 	/// <param name="source">The source data to use.</param>
-	/// <param name="capacity">The optional bounded capacity of the channel. Default is unbound.</param>
+	/// <param name="capacity">The optional bounded capacity of the channel.</param>
 	/// <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>
 	/// <param name="deferredExecution">If true, calls await Task.Yield() before writing to the channel.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, int, bool, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannel<T>(this IAsyncEnumerable<T> source,
 		int capacity, bool singleReader,
 		CancellationToken cancellationToken,
@@ -380,12 +414,7 @@ static async IAsyncEnumerable<TOut> AsAsyncEnumerableCore(Channel<TIn, TOut> cha
 	/// <summary>
 	/// Asynchronously executes all entries and writes their results to a channel.
 	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
-	/// <param name="source">The asynchronous source data to use.</param>
-	/// <param name="channelOptions">The options for configuring the new channel.</param>
-	/// <param name="maxConcurrency">The maximum number of concurrent operations.  Greater than 1 may likely cause results to be out of order.</param>
-	/// <param name="cancellationToken">An optional cancellation token.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, ChannelOptions, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannelAsync<T>(this IEnumerable<Func<T>> source,
 		ChannelOptions channelOptions, int maxConcurrency = 1,
 		CancellationToken cancellationToken = default)
@@ -395,28 +424,18 @@ public static ChannelReader<T> ToChannelAsync<T>(this IEnumerable<Func<T>> sourc
 	/// <summary>
 	/// Asynchronously executes all entries and writes their results to a channel.
 	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
-	/// <param name="source">The asynchronous source data to use.</param>
-	/// <param name="capacity">The optional bounded capacity of the channel. Default is unbound.</param>
-	/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
-	/// <param name="maxConcurrency">The maximum number of concurrent operations.  Greater than 1 may likely cause results to be out of order.</param>
-	/// <param name="cancellationToken">An optional cancellation token.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, int, bool, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannelAsync<T>(this IEnumerable<Func<T>> source,
 		int capacity = -1, bool singleReader = false, int maxConcurrency = 1,
 		CancellationToken cancellationToken = default)
 		=> CreateChannel<T>(capacity, singleReader)
 			.SourceAsync(maxConcurrency, source, cancellationToken);
 
-	/// <summary>
-	/// Writes all entries from the source to a channel and calls complete when finished.
-	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
 	/// <param name="source">The asynchronous source data to use.</param>
 	/// <param name="channelOptions">The options for configuring the new channel.</param>
 	/// <param name="maxConcurrency">The maximum number of concurrent operations.  Greater than 1 may likely cause results to be out of order.</param>
 	/// <param name="cancellationToken">An optional cancellation token.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, int, bool, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannelAsync<T>(this IEnumerable<ValueTask<T>> source,
 		ChannelOptions channelOptions, int maxConcurrency = 1,
 		CancellationToken cancellationToken = default)
@@ -439,31 +458,14 @@ public static ChannelReader<T> ToChannelAsync<T>(this IEnumerable<ValueTask<T>>
 		=> CreateChannel<T>(capacity, singleReader)
 			.SourceAsync(maxConcurrency, source, cancellationToken);
 
-	/// <summary>
-	/// Writes all entries from the source to a channel and calls complete when finished.
-	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
-	/// <param name="source">The asynchronous source data to use.</param>
-	/// <param name="channelOptions">The options for configuring the new channel.</param>
-	/// <param name="maxConcurrency">The maximum number of concurrent operations.  Greater than 1 may likely cause results to be out of order.</param>
-	/// <param name="cancellationToken">An optional cancellation token.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, ChannelOptions, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannelAsync<T>(this IEnumerable<Task<T>> source,
 		ChannelOptions channelOptions, int maxConcurrency = 1,
 		CancellationToken cancellationToken = default)
 		=> CreateChannel<T>(channelOptions)
 			.SourceAsync(maxConcurrency, source, cancellationToken);
 
-	/// <summary>
-	/// Writes all entries from the source to a channel and calls complete when finished.
-	/// </summary>
-	/// <typeparam name="T">The input type of the channel.</typeparam>
-	/// <param name="source">The asynchronous source data to use.</param>
-	/// <param name="capacity">The optional bounded capacity of the channel. Default is unbound.</param>
-	/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
-	/// <param name="maxConcurrency">The maximum number of concurrent operations.  Greater than 1 may likely cause results to be out of order.</param>
-	/// <param name="cancellationToken">An optional cancellation token.</param>
-	/// <returns>The channel reader containing the results.</returns>
+	/// <inheritdoc cref="ToChannelAsync{T}(IEnumerable{ValueTask{T}}, int, bool, int, CancellationToken)"/>
 	public static ChannelReader<T> ToChannelAsync<T>(this IEnumerable<Task<T>> source,
 		int capacity = -1, bool singleReader = false, int maxConcurrency = 1,
 		CancellationToken cancellationToken = default)