Fixes for join waiting.

Author: Oren (electricessence)

Open.ChannelExtensions.Tests/Program.cs

@@ -31,9 +31,9 @@ static async Task Main()
 				Console.WriteLine("Batch + join test 1...");
 				var sw = Stopwatch.StartNew();
 				var range = Enumerable
-					.Range(0, 2000000);
+					.Range(0, 10000000);
 
-				var result = new List<int>(2000000);
+				var result = new List<int>(10000000);
 
 				var total = await range
 					.ToChannel()
@@ -51,9 +51,9 @@ static async Task Main()
 				Console.WriteLine("Batch + join test 2...");
 				var sw = Stopwatch.StartNew();
 				var range = Enumerable
-					.Range(0, 2000000);
+					.Range(0, 10000000);
 
-				var result = new List<int>(2000000);
+				var result = new List<int>(10000000);
 
 				var total = await range
 					.ToChannel()

Open.ChannelExtensions/Extensions.Join.cs

@@ -17,23 +17,25 @@ public JoiningChannelReader(ChannelReader<TList> source)
 			{
 				_source = source ?? throw new ArgumentNullException(nameof(source));
 				Contract.EndContractBlock();
-			}
 
-			private readonly ChannelReader<TList> _source;
-			public override Task Completion
-				=> _source.Completion.ContinueWith(t =>
+				_source.Completion.ContinueWith(t =>
 				{
 					// Need to be sure writing is done before we continue...
 					lock (_buffer)
 					{
 						_buffer.Writer.Complete(t.Exception);
 					}
-					return _buffer.Reader.Completion;
-				})
-				.Unwrap();
+				});
+			}
+
+			private readonly ChannelReader<TList> _source;
+			public override Task Completion => _buffer.Reader.Completion;
 
 			bool TryPipeItems()
 			{
+				if (_source.Completion.IsCompleted)
+					return false;
+
 				lock (_buffer)
 				{
 					if (!_source.TryRead(out TList batch))
@@ -72,8 +74,8 @@ public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationTo
 					return b;
 
 				var s = _source.WaitToReadAsync(cancellationToken);
-				if (s.IsCompletedSuccessfully && !s.Result)
-					return b;
+				if (s.IsCompletedSuccessfully)
+					return s.Result ? new ValueTask<bool>(true) : b;
 
 				return WaitCore();