Cleanup.

Author: Oren (electricessence)

Open.ChannelExtensions/BufferingChannelReader.cs

@@ -8,7 +8,7 @@ namespace Open.ChannelExtensions
 {
 	abstract class BufferingChannelReader<TIn, TOut> : ChannelReader<TOut>
 	{
-		protected readonly ChannelReader<TIn> Source;
+		protected ChannelReader<TIn>? Source;
 		protected readonly Channel<TOut> Buffer;
 		public BufferingChannelReader(ChannelReader<TIn> source, bool singleReader)
 		{
@@ -17,20 +17,22 @@ public BufferingChannelReader(ChannelReader<TIn> source, bool singleReader)
 
 			Buffer = Extensions.CreateChannel<TOut>(-1, singleReader);
 
-			if (Source.Completion.IsCompleted)
+			if (source.Completion.IsCompleted)
 			{
-				Buffer.Writer.Complete(Source.Completion.Exception);
+				Buffer.Writer.Complete(source.Completion.Exception);
 			}
 			else
 			{
-				Source.Completion.ContinueWith(t =>
+				source.Completion.ContinueWith(t =>
 				{
 					// Need to be sure writing is done before we continue...
 					lock (Buffer)
 					{
 						while (TryPipeItems()) { }
 						Buffer.Writer.Complete(t.Exception);
 					}
+
+					Source = null;
 				});
 			}
 		}
@@ -59,11 +61,12 @@ public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationTo
 			if (cancellationToken.IsCancellationRequested)
 				return new ValueTask<bool>(Task.FromCanceled<bool>(cancellationToken));
 
+			var source = Source;
 			var b = Buffer.Reader.WaitToReadAsync(cancellationToken);
-			if (b.IsCompletedSuccessfully || Source.Completion.IsCompleted)
+			if (b.IsCompletedSuccessfully || source==null || source.Completion.IsCompleted)
 				return b;
 
-			var s = Source.WaitToReadAsync(cancellationToken);
+			var s = source.WaitToReadAsync(cancellationToken);
 			if (s.IsCompletedSuccessfully)
 				return s.Result ? new ValueTask<bool>(true) : b;
 

Open.ChannelExtensions/Extensions.Batch.cs

@@ -39,7 +39,8 @@ protected override bool TryPipeItems()
 					if (Buffer.Reader.Completion.IsCompleted)
 						return false;
 
-					if (Source.Completion.IsCompleted)
+					var source = Source;
+					if (source == null || source.Completion.IsCompleted)
 					{
 						c.TrimExcess();
 						_current = null;
@@ -50,7 +51,7 @@ protected override bool TryPipeItems()
 						return true;
 					}
 
-					while (Source.TryRead(out T item))
+					while (source.TryRead(out T item))
 					{
 						if (c.Count == _batchSize)
 						{

Open.ChannelExtensions/Extensions.Join.cs

@@ -17,12 +17,13 @@ public JoiningChannelReader(ChannelReader<TList> source, bool singleReader) : ba
 
 			protected override bool TryPipeItems()
 			{
-				if (Source.Completion.IsCompleted)
+				var source = Source;
+				if (source==null || source.Completion.IsCompleted)
 					return false;
 
 				lock (Buffer)
 				{
-					if (!Source.TryRead(out TList batch))
+					if (!source.TryRead(out TList batch))
 						return false;
 
 					foreach (var i in batch)
@@ -36,34 +37,6 @@ protected override bool TryPipeItems()
 			}
 		}
 
-		const int DefaultBufferSize = 100;
-
-		static ChannelReader<T> JoinInternal<T, TEnumerable>(ChannelReader<TEnumerable> source, int bufferSize, bool singleReader)
-			where TEnumerable : IEnumerable<T>
-		{
-			if (source == null) throw new ArgumentNullException(nameof(source));
-			if (bufferSize < 1) throw new ArgumentOutOfRangeException(nameof(bufferSize), bufferSize, "Must be at least 1.");
-			Contract.EndContractBlock();
-
-			var buffer = CreateChannel<T>(bufferSize, singleReader);
-			var writer = buffer.Writer;
-
-			Task.Run(async () =>
-				await source.ReadAllAsync(
-					async (batch, i) =>
-					{
-						foreach (var e in batch)
-						{
-							if(!writer.TryWrite(e))
-								await writer.WriteAsync(e).ConfigureAwait(false);
-						}
-					}))
-				.ContinueWith(
-					t => buffer.CompleteAsync(t.Exception), TaskContinuationOptions.ExecuteSynchronously);
-
-			return buffer.Reader;
-		}
-
 		/// <summary>
 		/// Joins collections of the same type into a single channel reader in the order provided.
 		/// </summary>
@@ -123,7 +96,7 @@ public static ChannelReader<T> Join<T>(this ChannelReader<T[]> source, bool sing
 		/// <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 = DefaultBufferSize, bool singleReader = false)
+		public static ChannelReader<T> Join<T>(this ChannelReader<IAsyncEnumerable<T>> source, int bufferSize = 100, bool singleReader = false)
 		{
 			var buffer = CreateChannel<T>(bufferSize, singleReader);