Added Merge extension with tests.

Author: electricessence

Open.ChannelExtensions.Tests/MergeTests.cs

@@ -0,0 +1,68 @@
+namespace Open.ChannelExtensions.Tests;
+
+public static class MergeTests
+{
+	[Fact()]
+	public static async Task BasicMergeTest()
+	{
+		// Arrange
+		const int total = 3000000;
+		const int bound = 100;
+
+		// 3 channels
+		var c1 = Channel.CreateBounded<int>(bound);
+		var c2 = Channel.CreateBounded<int>(bound);
+		var c3 = Channel.CreateBounded<int>(bound);
+		var writers = new[] { c1.Writer, c2.Writer, c3.Writer };
+
+		// 3 readers
+		var merging = new[] { c1.Reader, c2.Reader, c3.Reader }.Merge().ToListAsync(total);
+
+		// Act
+		await Parallel.ForAsync(0, total,
+			(i, token) => writers[i % 3].WriteAsync(i, token));
+
+		foreach (var writer in writers)
+			writer.Complete();
+
+		var merged = await merging;
+		merged.Sort();
+
+		// Assert
+		Assert.Equal(total, merged.Count);
+		Assert.True(Enumerable.Range(0, total).SequenceEqual(merged));
+	}
+
+	[Fact()]
+	public static async Task ExceptionPropagationTest()
+	{
+		// Arrange
+		const int total = 3000000;
+		const int bound = 100;
+
+		// 3 channels
+		var c1 = Channel.CreateBounded<int>(bound);
+		var c2 = Channel.CreateBounded<int>(bound);
+		var c3 = Channel.CreateBounded<int>(bound);
+		var writers = new[] { c1.Writer, c2.Writer, c3.Writer };
+
+		// 3 readers
+		var merging = new[] { c1.Reader, c2.Reader, c3.Reader }.Merge();
+		var list = merging.ToListAsync(total);
+
+		// Act
+		await Assert.ThrowsAsync<ChannelClosedException>(() => Parallel.ForAsync(0, total,
+			async (i, token) =>
+			{
+				var w = writers[i % 3];
+				if (i == total / 2)
+					w.Complete(new Exception("Test"));
+				else
+					await w.WriteAsync(i, token).ConfigureAwait(false);
+			}));
+
+		// Assert
+		await Assert.ThrowsAsync<Exception>(list.AsTask);
+		await Assert.ThrowsAsync<Exception>(() => merging.Completion);
+	}
+}

Open.ChannelExtensions/Extensions.Merge.cs

@@ -0,0 +1,118 @@
+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)
+		{
+			// 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;
+				}
+			}
+
+			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.
+	/// </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);
+}

Open.ChannelExtensions/Extensions.Read.cs

@@ -1058,13 +1058,14 @@ public static ValueTask<long> ReadAllAsLines<T>(this Channel<T, string> channel,
 	/// </summary>
 	/// <typeparam name="T">The item type.</typeparam>
 	/// <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)
+	public static async ValueTask<List<T>> ToListAsync<T>(this ChannelReader<T> reader, int initialCapacity = 0)
 	{
 		if (reader is null) throw new ArgumentNullException(nameof(reader));
 		Contract.EndContractBlock();
 
-		var list = new List<T>();
+		List<T> list = initialCapacity == 0 ? new() : new(initialCapacity);
 		await ReadAll(reader, list.Add).ConfigureAwait(false);
 		return list;
 	}