Added ForceBatch() to BatchChannelReader.

Extracted buffering and batch readers to extend functionality.

Author: Oren (electricessence)

Open.ChannelExtensions/BatchingChannelReader.cs

@@ -0,0 +1,95 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Contracts;
+using System.Threading.Channels;
+
+namespace Open.ChannelExtensions
+{
+	/// <summary>
+	/// A ChannelReader that batches results.
+	/// Use the .Batch extension instead of constructing this directly.
+	/// </summary>
+	public class BatchingChannelReader<T> : BufferingChannelReader<T, List<T>>
+	{
+		private readonly int _batchSize;
+		private List<T>? _current;
+
+		/// <summary>
+		/// Constructs a BatchingChannelReader.
+		/// Use the .Batch extension instead of constructing this directly.
+		/// </summary>
+		public BatchingChannelReader(ChannelReader<T> source, int batchSize, bool singleReader, bool syncCont = false) : base(source, singleReader, syncCont)
+		{
+			if (batchSize < 1) throw new ArgumentOutOfRangeException(nameof(batchSize), batchSize, "Must be at least 1.");
+			Contract.EndContractBlock();
+
+			_batchSize = batchSize;
+			_current = source.Completion.IsCompleted ? null : new List<T>(batchSize);
+		}
+
+		/// <summary>
+		/// If no full batch is waiting, will force buffering any batch that has at least one item.
+		/// Returns true if anything was added to the buffer.
+		/// </summary>
+		public bool ForceBatch()
+		{
+			if (Buffer == null || Buffer.Reader.Completion.IsCompleted) return false;
+			if (TryPipeItems()) return true;
+
+			lock (Buffer)
+			{
+				if (Buffer.Reader.Completion.IsCompleted) return false;
+				if (TryPipeItems()) return true;
+				var c = _current;
+				if (c == null || c.Count == 0 || Buffer.Reader.Completion.IsCompleted)
+					return false;
+				c.TrimExcess();
+				_current = new List<T>(_batchSize);
+				Buffer.Writer.TryWrite(c);
+			}
+
+			return true;
+		}
+
+		/// <inheritdoc />
+		protected override bool TryPipeItems()
+		{
+			if (_current == null || Buffer == null || Buffer.Reader.Completion.IsCompleted)
+				return false;
+
+			lock (Buffer)
+			{
+				var c = _current;
+				if (c == null || Buffer.Reader.Completion.IsCompleted)
+					return false;
+
+				var source = Source;
+				if (source == null || source.Completion.IsCompleted)
+				{
+					// All finished, release the last batch to the buffer.
+					c.TrimExcess();
+					_current = null;
+					if (c.Count == 0)
+						return false;
+
+					Buffer.Writer.TryWrite(c);
+					return true;
+				}
+
+				while (source.TryRead(out T item))
+				{
+					if (c.Count == _batchSize)
+					{
+						_current = new List<T>(_batchSize) { item };
+						Buffer.Writer.TryWrite(c);
+						return true;
+					}
+
+					c.Add(item);
+				}
+
+				return false;
+			}
+		}
+	}
+}

Open.ChannelExtensions/BufferingChannelReader.cs

@@ -6,11 +6,28 @@
 
 namespace Open.ChannelExtensions
 {
-	abstract class BufferingChannelReader<TIn, TOut> : ChannelReader<TOut>
+	/// <summary>
+	/// Base class for buffering results of a source ChannelReader.
+	/// </summary>
+	/// <typeparam name="TIn">The input type of the buffer.</typeparam>
+	/// <typeparam name="TOut">The output type of the buffer.</typeparam>
+	public abstract class BufferingChannelReader<TIn, TOut> : ChannelReader<TOut>
 	{
-		protected ChannelReader<TIn>? Source;
-		protected readonly Channel<TOut>? Buffer;
-		public BufferingChannelReader(ChannelReader<TIn> source, bool singleReader, bool syncCont = false)
+		/// <summary>
+		/// The source of the buffer.
+		/// </summary>
+		protected ChannelReader<TIn>? Source { get; set; }
+
+		/// <summary>
+		/// The internal channel used for buffering.
+		/// </summary>
+		protected Channel<TOut>? Buffer { get; }
+
+
+		/// <summary>
+		/// Base constructor for a BufferingChannelReader.
+		/// </summary>
+		protected BufferingChannelReader(ChannelReader<TIn> source, bool singleReader, bool syncCont = false)
 		{
 			Source = source ?? throw new ArgumentNullException(nameof(source));
 			Contract.EndContractBlock();
@@ -37,10 +54,16 @@ public BufferingChannelReader(ChannelReader<TIn> source, bool singleReader, bool
 			}
 		}
 
+		/// <inheritdoc />
 		public override Task Completion => Buffer?.Reader.Completion ?? Task.CompletedTask;
 
+		/// <summary>
+		/// The method that triggers adding entries to the buffer.
+		/// </summary>
+		/// <returns></returns>
 		protected abstract bool TryPipeItems();
 
+		/// <inheritdoc />
 		public override bool TryRead(out TOut item)
 		{
 			if (Buffer != null) do
@@ -54,6 +77,7 @@ public override bool TryRead(out TOut item)
 			return false;
 		}
 
+		/// <inheritdoc />
 		public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
 		{
 			if (Buffer == null || Buffer.Reader.Completion.IsCompleted)