Updates. (checkpoint)

Author: electricessence

.editorconfig

@@ -117,8 +117,8 @@ csharp_style_prefer_index_operator = true:silent
 csharp_style_prefer_null_check_over_type_check = true:warning
 csharp_style_prefer_range_operator = true:silent
 csharp_style_throw_expression = true:warning
-csharp_style_unused_value_assignment_preference = discard_variable
-csharp_style_unused_value_expression_statement_preference = discard_variable
+csharp_style_unused_value_assignment_preference = discard_variable:suggestion
+csharp_style_unused_value_expression_statement_preference = discard_variable:silent
 
 # 'using' directive preferences
 csharp_using_directive_placement = outside_namespace:warning
@@ -236,10 +236,11 @@ dotnet_style_null_propagation = true:warning
 dotnet_style_prefer_is_null_check_over_reference_equality_method = true:warning
 dotnet_style_prefer_auto_properties = true:suggestion
 dotnet_style_object_initializer = true:suggestion
-dotnet_style_prefer_collection_expression = true:suggestion
+dotnet_style_prefer_collection_expression = when_types_exactly_match:suggestion
 dotnet_style_collection_initializer = true:suggestion
 dotnet_style_prefer_simplified_boolean_expressions = true:warning
 dotnet_style_prefer_conditional_expression_over_assignment = true:warning
 dotnet_style_prefer_conditional_expression_over_return = true:warning
 dotnet_style_explicit_tuple_names = true:suggestion
-dotnet_diagnostic.CA1510.severity = none
+dotnet_diagnostic.CA1510.severity = none
+dotnet_code_quality_unused_parameters = all:suggestion

Open.ChannelExtensions/BatchingChannelReader.cs

@@ -246,12 +246,9 @@ protected override async ValueTask<bool> WaitToReadAsyncCore(
 }
 
 /// <inheritdoc />
-public class QueueBatchingChannelReader<T> : BatchingChannelReader<T, Queue<T>>
+public class QueueBatchingChannelReader<T>(ChannelReader<T> source, int batchSize, bool singleReader, bool syncCont = false)
+	: BatchingChannelReader<T, Queue<T>>(source, batchSize, singleReader, syncCont)
 {
-	/// <inheritdoc />
-	public QueueBatchingChannelReader(ChannelReader<T> source, int batchSize, bool singleReader, bool syncCont = false)
-		: base(source, batchSize, singleReader, syncCont) { }
-
 	/// <inheritdoc />
 	[MethodImpl(MethodImplOptions.AggressiveInlining)]
 	protected override void AddBatchItem(Queue<T> batch, T item)
@@ -282,12 +279,9 @@ protected override void TrimBatch(Queue<T> batch)
 }
 
 /// <inheritdoc />
-public class BatchingChannelReader<T> : BatchingChannelReader<T, List<T>>
+public class BatchingChannelReader<T>(ChannelReader<T> source, int batchSize, bool singleReader, bool syncCont = false)
+	: BatchingChannelReader<T, List<T>>(source, batchSize, singleReader, syncCont)
 {
-	/// <inheritdoc />
-	public BatchingChannelReader(ChannelReader<T> source, int batchSize, bool singleReader, bool syncCont = false)
-		: base(source, batchSize, singleReader, syncCont) { }
-
 	/// <inheritdoc />
 	[MethodImpl(MethodImplOptions.AggressiveInlining)]
 	protected override void AddBatchItem(List<T> batch, T item)

Open.ChannelExtensions/Extensions.Join.cs

@@ -2,13 +2,10 @@
 
 public static partial class Extensions
 {
-	sealed class JoiningChannelReader<TList, T> : BufferingChannelReader<TList, T>
+	sealed class JoiningChannelReader<TList, T>(ChannelReader<TList> source, bool singleReader)
+		: BufferingChannelReader<TList, T>(source, singleReader)
 		where TList : IEnumerable<T>
 	{
-		public JoiningChannelReader(ChannelReader<TList> source, bool singleReader) : base(source, singleReader)
-		{
-		}
-
 		protected override bool TryPipeItems(bool _)
 		{
 			ChannelReader<TList>? source = Source;

Open.ChannelExtensions/Extensions.Merge.cs

@@ -1,128 +1,47 @@
-namespace Open.ChannelExtensions;
+using System.Collections.Immutable;
+
+namespace Open.ChannelExtensions;
 
 public static partial class Extensions
 {
-	sealed class MergingChannelReader<T> : ChannelReader<T>
-	{
-		public MergingChannelReader(IEnumerable<ChannelReader<T>> sources)
-		{
-			if (sources is null) throw new ArgumentNullException(nameof(sources));
-			Contract.EndContractBlock();
-
-			var active = sources.Where(s =>
-			{
-				Debug.Assert(s is not null);
-				return s.Completion.Status != TaskStatus.RanToCompletion;
-			}).ToArray();
-
-			_count = active.Length;
-
-			if (_count == 0)
-			{
-				_sources = [];
-				Completion = Task.CompletedTask;
-				return;
-			}
-
-			_sources = active;
-
-			// Capture the initial list of completions.
-			var completions = active.Select(e => e.Completion).ToList();
-
-			// Create a task that completes when any of the sources are faulted:
-			Completion = Task.Run(async () =>
-			{
-				// Wait for any of the tasks to complete and let it throw if it faults.
-				while (completions.Count != 0)
-				{
-					var completed = await Task.WhenAny(completions).ConfigureAwait(false);
-					// Propagate the exception.
-					await completed.ConfigureAwait(false);
-					completions.Remove(completed);
-				}
-			});
-		}
-
-		private readonly ChannelReader<T>[] _sources;
-		public override Task Completion { get; }
-
-		readonly int _count;
-		int _next = -1;
-
-		public override bool TryRead(out T item)
-		{
-			int previous = -1;
-			// Try as many times as there are sources before giving up.
-			for (var attempt = 0; attempt < _count; attempt++)
-			{
-				// If the value overflows, it will be negative, which is fine, we'll adapt.
-				var i = Interlocked.Increment(ref _next) % _count;
-				if (i < 0) i += _count;
-
-				var source = _sources[i];
-
-				if (source.TryRead(out T? s))
-				{
-					item = s;
-					return true;
-				}
-
-				// Help the round-robin to try each source at least once.
-				// If previous is not -1 and i is not the next in the sequence,
-				// then another thread has already tried that source.
-				if (previous != -1 && (previous + 1) % _count != i)
-					attempt--; // Allow for an extra attempt. 
-
-				previous = i;
-			}
-
-			item = default!;
-			return false;
-		}
-
-		public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
-		{
-			var completion = Completion;
-			if (Completion.IsCompleted)
-			{
-				return completion.IsFaulted
-					? new ValueTask<bool>(Task.FromException<bool>(completion.Exception!))
-					: new ValueTask<bool>(false);
-			}
-
-			if (cancellationToken.IsCancellationRequested)
-				return new ValueTask<bool>(Task.FromCanceled<bool>(cancellationToken));
-
-			// Not complete or cancelled?  Wait on the sources.
-			return WaitToReadAsyncCore(cancellationToken);
-		}
-
-		private async ValueTask<bool> WaitToReadAsyncCore(CancellationToken cancellationToken)
-		{
-		retry:
-			// We don't care about ones that have already completed.
-			var active = _sources.Where(s => s.Completion.Status != TaskStatus.RanToCompletion).ToArray();
-			if (active.Length == 0) return false;
-
-			var next = await Task.WhenAny(active.Select(s => s.WaitToReadAsync(cancellationToken).AsTask())).ConfigureAwait(false);
-
-			// Allow for possible exception to be thrown.
-			var result = await next.ConfigureAwait(false);
-			if (result) return true;
-
-			// If result was false, then there's one less and we should try again.
-			goto retry;
-		}
-	}
-
 	/// <summary>
-	/// Reads from multiple sources in a round-robin fashion.
+	/// Creates a <see cref="MergingChannelReader{T}"/>
+	/// that reads from multiple sources in a round-robin fashion.
 	/// </summary>
 	/// <typeparam name="T">The source type.</typeparam>
 	/// <param name="sources">The channels to read from.</param>
-	/// <returns>
-	/// A <see cref="ChannelReader{T}"/> that reads from all sources in a round-robin fashion.
-	/// </returns>
-	public static ChannelReader<T> Merge<T>(this IEnumerable<ChannelReader<T>> sources)
-		=> new MergingChannelReader<T>(sources);
+	public static MergingChannelReader<T> Merge<T>(this IEnumerable<ChannelReader<T>> sources) => new(sources);
+
+	/// <summary>
+	/// Merges the <paramref name="primary"/> with the <paramref name="secondary"/>
+	/// as a <see cref="MergingChannelReader{T}"/>
+	/// that reads from multiple sources in a round-robin fashion.
+	/// </summary>
+	/// <exception cref="ArgumentNullException">
+	/// If the <paramref name="primary"/>
+	/// or <paramref name="secondary"/> sources are null.
+	/// </exception>
+	/// <inheritdoc cref="MergingChannelReader{T}.Merge(ChannelReader{T}, ChannelReader{T}[])"/>/>
+	public static MergingChannelReader<T> Merge<T>(
+		this ChannelReader<T> primary,
+		ChannelReader<T> secondary,
+		params ChannelReader<T>[] others)
+	{
+		if (primary is null) throw new ArgumentNullException(nameof(primary));
+		if (secondary is null) throw new ArgumentNullException(nameof(secondary));
+		Contract.EndContractBlock();
+
+		// Is this already a merging reader? Then recapture the sources so it flattens the hierarchy.
+		if (primary is MergingChannelReader<T> mcr)
+			return mcr.Merge(secondary, others);
+
+		if(others is null || others.Length == 0)
+			return new MergingChannelReader<T>(ImmutableArray.Create(primary, secondary));
+
+		var builder = ImmutableArray.CreateBuilder<ChannelReader<T>>(2 + others.Length);
+		builder.Add(primary);
+		builder.Add(secondary);
+		builder.AddRange(others);
+		return new MergingChannelReader<T>(builder.MoveToImmutable());
+	}
 }

Open.ChannelExtensions/Extensions.Read.cs

@@ -1060,12 +1060,12 @@ public static ValueTask<long> ReadAllAsLines<T>(this Channel<T, string> channel,
 	/// <param name="reader">The channel reader to read from.</param>
 	/// <param name="initialCapacity">An optional capacity to initialze the list with.</param>
 	/// <returns>A list containing all the items from the completed channel.</returns>
-	public static async ValueTask<List<T>> ToListAsync<T>(this ChannelReader<T> reader, int initialCapacity = 0)
+	public static async ValueTask<List<T>> ToListAsync<T>(this ChannelReader<T> reader, int initialCapacity = -1)
 	{
 		if (reader is null) throw new ArgumentNullException(nameof(reader));
 		Contract.EndContractBlock();
 
-		List<T> list = initialCapacity == 0 ? new() : new(initialCapacity);
+		List<T> list = initialCapacity < 0 ? new() : new(initialCapacity);
 		await ReadAll(reader, list.Add).ConfigureAwait(false);
 		return list;
 	}