Final touches before release.

Author: Oren (electricessence)

Open.ChannelExtensions.Tests/BasicTests.cs

@@ -121,8 +121,8 @@ public static async Task BatchJoin(int testSize, int batchSize)
 			var sw = Stopwatch.StartNew();
 			var total = await range
 				.ToChannel(singleReader: true)
-				.Batch(batchSize)
-				.Join()
+				.Batch(batchSize, singleReader: true)
+				.Join(singleReader: true)
 				.ReadAll(i => result.Add(i));
 			sw.Stop();
 
@@ -133,6 +133,40 @@ public static async Task BatchJoin(int testSize, int batchSize)
 			Assert.True(result.SequenceEqual(range));
 		}
 
+		[Theory]
+		[InlineData(11)]
+		[InlineData(51)]
+		[InlineData(101)]
+		[InlineData(1001)]
+		public static async Task JoinAsync(int repeat)
+		{
+			var testSize = repeat * repeat;
+			var range = Enumerable.Repeat(Samples(), repeat);
+			var result = new List<int>(testSize);
+
+			var sw = Stopwatch.StartNew();
+			var total = await range
+				.ToChannel(singleReader: true)
+				.Join(singleReader: true)
+				.ReadAll(i => result.Add(i));
+			sw.Stop();
+
+			Console.WriteLine("Channel<IAsyncEnumerable>.Join(): {0}", sw.Elapsed);
+			Console.WriteLine();
+
+			Assert.Equal(testSize, result.Count);
+			Assert.True(result.SequenceEqual(Enumerable.Repeat(Enumerable.Range(0, repeat), repeat).SelectMany(i => i)));
+
+			async IAsyncEnumerable<int> Samples()
+			{
+				for (var i = 0; i < repeat; i++)
+				{
+					var x = new ValueTask<int>(i);
+					yield return await x;
+				}
+			}
+		}
+
 		[Theory]
 		[InlineData(testSize1)]
 		[InlineData(testSize2)]

Open.ChannelExtensions/BufferingChannelReader.cs

@@ -9,20 +9,20 @@ namespace Open.ChannelExtensions
 	abstract class BufferingChannelReader<TIn, TOut> : ChannelReader<TOut>
 	{
 		protected ChannelReader<TIn>? Source;
-		protected readonly Channel<TOut> Buffer;
+		protected readonly Channel<TOut>? Buffer;
 		public BufferingChannelReader(ChannelReader<TIn> source, bool singleReader)
 		{
 			Source = source ?? throw new ArgumentNullException(nameof(source));
 			Contract.EndContractBlock();
 
-			Buffer = Extensions.CreateChannel<TOut>(-1, singleReader);
-
 			if (source.Completion.IsCompleted)
 			{
-				Buffer.Writer.Complete(source.Completion.Exception);
+				Buffer = null;
 			}
 			else
 			{
+				Buffer = Extensions.CreateChannel<TOut>(-1, singleReader);
+
 				source.Completion.ContinueWith(t =>
 				{
 					// Need to be sure writing is done before we continue...
@@ -37,55 +37,58 @@ public BufferingChannelReader(ChannelReader<TIn> source, bool singleReader)
 			}
 		}
 
-		public override Task Completion => Buffer.Reader.Completion;
+		public override Task Completion => Buffer?.Reader.Completion ?? Task.CompletedTask;
 
 		protected abstract bool TryPipeItems();
 
 		public override bool TryRead(out TOut item)
 		{
-			do
+			if (Buffer != null) do
 			{
 				if (Buffer.Reader.TryRead(out item))
 					return true;
 			}
 			while (TryPipeItems());
 
-#pragma warning disable CS8653 // A default expression introduces a null value for a type parameter.
-			item = default;
-#pragma warning restore CS8653 // A default expression introduces a null value for a type parameter.
+			item = default!;
 			return false;
 		}
 
 		public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
 		{
+			if (Buffer == null || Buffer.Reader.Completion.IsCompleted)
+				return new ValueTask<bool>(false);
+
 			if (cancellationToken.IsCancellationRequested)
 				return new ValueTask<bool>(Task.FromCanceled<bool>(cancellationToken));
 
-			var source = Source;
 			var b = Buffer.Reader.WaitToReadAsync(cancellationToken);
-			if (b.IsCompletedSuccessfully || source==null || source.Completion.IsCompleted)
+			if (b.IsCompleted)
 				return b;
 
-			var s = source.WaitToReadAsync(cancellationToken);
-			if (s.IsCompletedSuccessfully)
-				return s.Result ? new ValueTask<bool>(true) : b;
+			var source = Source;
+			if (source == null)
+				return b;
 
 			return WaitCore();
 
 			async ValueTask<bool> WaitCore()
 			{
-				cancellationToken.ThrowIfCancellationRequested();
-
-				// Not sure if there's a better way to 'WhenAny' with a ValueTask yet.
-				var bt = b.AsTask();
-				var st = s.AsTask();
-				var first = await Task.WhenAny(bt, st);
-				// Either one? Ok go.
-				if (first.Result) return true;
-				// Buffer returned false? We're done.
-				if (first == bt) return false;
-				// Second return false? Wait for buffer.
-				return await bt;
+
+			start:
+
+				if (b.IsCompleted) return await b;
+
+				var s = source.WaitToReadAsync(cancellationToken);
+				if (s.IsCompleted && !b.IsCompleted)
+					TryPipeItems();
+
+				if (b.IsCompleted) return await b;
+				await s;
+				if (b.IsCompleted) return await b;
+				TryPipeItems();
+
+				goto start;
 			}
 		}
 	}

Open.ChannelExtensions/Documentation.xml

@@ -21,10 +21,9 @@ public BatchingChannelReader(ChannelReader<T> source, int batchSize, bool single
 				_current = source.Completion.IsCompleted ? null : new List<T>(batchSize);
 			}
 
-
 			protected override bool TryPipeItems()
 			{
-				if (_current == null)
+				if (_current == null || Buffer == null)
 					return false;
 
 				if (Buffer.Reader.Completion.IsCompleted)

Open.ChannelExtensions/Extensions.Batch.cs

@@ -29,9 +29,7 @@ public override bool TryRead(out T item)
 						return true;
 				}
 
-#pragma warning disable CS8653 // A default expression introduces a null value for a type parameter.
-				item = default;
-#pragma warning restore CS8653 // A default expression introduces a null value for a type parameter.
+				item = default!;
 				return false;
 			}
 

Open.ChannelExtensions/Extensions.Filter.cs

@@ -16,7 +16,7 @@ public JoiningChannelReader(ChannelReader<TList> source, bool singleReader) : ba
 			protected override bool TryPipeItems()
 			{
 				var source = Source;
-				if (source==null || source.Completion.IsCompleted)
+				if (source == null || source.Completion.IsCompleted || Buffer == null)
 					return false;
 
 				lock (Buffer)
@@ -91,20 +91,20 @@ public static ChannelReader<T> Join<T>(this ChannelReader<T[]> source, bool sing
 		/// </summary>
 		/// <typeparam name="T">The result type.</typeparam>
 		/// <param name="source">The source reader.</param>
-		/// <param name="bufferSize">The capacity of the resultant channel.</param>
 		/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
 		/// <returns>A channel reader containing the joined results.</returns>
-		public static ChannelReader<T> Join<T>(this ChannelReader<IAsyncEnumerable<T>> source, int bufferSize = 100, bool singleReader = false)
+		public static ChannelReader<T> Join<T>(this ChannelReader<IAsyncEnumerable<T>> source, bool singleReader = false)
 		{
-			var buffer = CreateChannel<T>(bufferSize, singleReader);
+			var buffer = CreateChannel<T>(1, singleReader);
 
-			Task.Run(async () =>
-				await source.ReadAllAsync(
+			source
+				.ReadAllAsync(
 					async (batch, i) =>
 					{
 						await foreach (var e in batch)
 							await buffer.Writer.WriteAsync(e).ConfigureAwait(false);
-					}))
+					})
+				.AsTask()
 				.ContinueWith(
 					t => buffer.CompleteAsync(t.Exception), TaskContinuationOptions.ExecuteSynchronously);
 

Open.ChannelExtensions/Extensions.Join.cs

@@ -29,9 +29,7 @@ public override bool TryRead(out TOut item)
 					return true;
 				}
 
-#pragma warning disable CS8653 // A default expression introduces a null value for a type parameter.
-				item = default;
-#pragma warning restore CS8653 // A default expression introduces a null value for a type parameter.
+				item = default!;
 				return false;
 			}
 

Open.ChannelExtensions/Extensions.Transform.cs

@@ -30,9 +30,7 @@ public override bool TryRead(out T item)
 					}
 				}
 
-#pragma warning disable CS8653 // A default expression introduces a null value for a type parameter.
-				item = default;
-#pragma warning restore CS8653 // A default expression introduces a null value for a type parameter.
+				item = default!;
 				return false;
 			}
 

Open.ChannelExtensions/Extensions.TypeFilter.cs

@@ -91,9 +91,7 @@ static bool TryMoveNextSynchronized<T>(IEnumerator<T> source, out T value)
 				}
 			}
 
-#pragma warning disable CS8653 // A default expression introduces a null value for a type parameter.
-			value = default;
-#pragma warning restore CS8653 // A default expression introduces a null value for a type parameter.
+			value = default!;
 			return false;
 		}
 

Open.ChannelExtensions/Extensions.WriteConcurrently.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
@@ -190,6 +191,48 @@ public static ChannelReader<T> ToChannel<T>(this IAsyncEnumerable<T> source,
 			bool deferredExecution = false)
 			=> CreateChannel<T>(capacity, singleReader)
 				.Source(source, cancellationToken, deferredExecution);
+
+		/// <summary>
+		/// Iterates over the results in a ChannelReader.
+		/// Provided as an alternative to .ReadAllAsync() which at the time of publishing this, only exists in .NET Core 3.0 and not .NET Standard 2.1
+		/// </summary>
+		/// <typeparam name="T">The output type of the channel.</typeparam>
+		/// <param name="reader">The reader to read from.</param>
+		/// <param name="cancellationToken">An optional cancellation token that will break out of the iteration.</param>
+		/// <returns>An IAsyncEnumerable for iterating the channel.</returns>
+		public static async IAsyncEnumerable<T> AsAsyncEnumerable<T>(this ChannelReader<T> reader, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+		{
+			do
+			{
+				while (!cancellationToken.IsCancellationRequested && reader.TryRead(out var item))
+					yield return item;
+			}
+			while (
+				!cancellationToken.IsCancellationRequested
+				&& await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false));
+		}
+
+		/// <summary>
+		/// Iterates over the results in a Channel.
+		/// Provided as an alternative to .ReadAllAsync() which at the time of publishing this, only exists in .NET Core 3.0 and not .NET Standard 2.1
+		/// </summary>
+		/// <typeparam name="TIn">The type recieved by the source channel.</typeparam>
+		/// <typeparam name="TOut">The outgoing type from the source channel.</typeparam>
+		/// <param name="channel">The reader to read from.</param>
+		/// <param name="cancellationToken">An optional cancellation token that will break out of the iteration.</param>
+		/// <returns>An IAsyncEnumerable for iterating the channel.</returns>
+		public static async IAsyncEnumerable<TOut> AsAsyncEnumerable<TIn, TOut>(this Channel<TIn, TOut> channel, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+		{
+			var reader = channel.Reader;
+			do
+			{
+				while (!cancellationToken.IsCancellationRequested && reader.TryRead(out var item))
+					yield return item;
+			}
+			while (
+				!cancellationToken.IsCancellationRequested
+				&& await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false));
+		}
 #endif
 
 		/// <summary>