Added tests. Expanded join options.

Author: Oren (electricessence)

Open.ChannelExtensions.Tests/Program.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
@@ -27,15 +28,41 @@ static async Task Main()
 			}
 
 			{
-				Console.WriteLine("Standard Channel operation test...");
+				Console.WriteLine("Batch + join test 1...");
 				var sw = Stopwatch.StartNew();
-				var total = await Enumerable
-					.Repeat((Func<int, ValueTask<int>>)Delay, repeat)
-					.Select((t, i) => t(i))
-					.ToChannelAsync(singleReader: true)
-					.ReadAll(Dummy);
+				var range = Enumerable
+					.Range(0, 2000000);
+
+				var result = new List<int>(2000000);
+
+				var total = await range
+					.ToChannel()
+					.Batch(5000)
+					.Join()
+					.ReadAll(i=> result.Add(i));
+
 				sw.Stop();
-				Debug.Assert(total == repeat);
+				Debug.Assert(result.SequenceEqual(range));
+				Console.WriteLine(sw.Elapsed);
+				Console.WriteLine();
+			}
+
+			{
+				Console.WriteLine("Batch + join test 2...");
+				var sw = Stopwatch.StartNew();
+				var range = Enumerable
+					.Range(0, 2000000);
+
+				var result = new List<int>(2000000);
+
+				var total = await range
+					.ToChannel()
+					.Batch(50)
+					.Join()
+					.ReadAll(i => result.Add(i));
+
+				sw.Stop();
+				Debug.Assert(result.SequenceEqual(range));
 				Console.WriteLine(sw.Elapsed);
 				Console.WriteLine();
 			}

Open.ChannelExtensions/Extensions.Join.cs

@@ -95,8 +95,20 @@ async ValueTask<bool> WaitCore()
 			}
 		}
 
-		public static ChannelReader<T> Join<TList, T>(this ChannelReader<TList> source)
-			where TList : IEnumerable<T>
-			=> new JoiningChannelReader<TList, T>(source);
+		public static ChannelReader<T> Join<T>(this ChannelReader<IEnumerable<T>> source)
+			=> new JoiningChannelReader<IEnumerable<T>, T>(source);
+
+		public static ChannelReader<T> Join<T>(this ChannelReader<ICollection<T>> source)
+			=> new JoiningChannelReader<ICollection<T>, T>(source);
+
+		public static ChannelReader<T> Join<T>(this ChannelReader<IList<T>> source)
+			=> new JoiningChannelReader<IList<T>, T>(source);
+
+		public static ChannelReader<T> Join<T>(this ChannelReader<List<T>> source)
+			=> new JoiningChannelReader<List<T>, T>(source);
+
+		public static ChannelReader<T> Join<T>(this ChannelReader<T[]> source)
+			=> new JoiningChannelReader<T[], T>(source);
+
 	}
 }