Previous test pass after adding timeout code.

Author: electricessence

Open.ChannelExtensions/BatchingChannelReader.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Diagnostics.Contracts;
 using System.Threading;
 using System.Threading.Channels;
@@ -20,7 +21,12 @@ public class BatchingChannelReader<T> : BufferingChannelReader<T, List<T>>
 	/// 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)
+	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();
@@ -32,31 +38,59 @@ public BatchingChannelReader(ChannelReader<T> source, int batchSize, bool single
 	/// 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()
+	public bool ForceBatch() => TryPipeItems(true);
+
+	long _timeout = -1;
+	Timer? _timer;
+
+	/// <summary>
+	/// Specifies a timeout by which a batch will be emmited there is at least one item but has been waiting
+	/// for longer than the timeout value.
+	/// </summary>
+	/// <param name="millisecondsTimeout">
+	/// The timeout value where after a batch is forced.<br/>
+	/// A value of zero or less cancels/clears any timeout.
+	/// </param>
+	/// <returns>The current reader.</returns>
+	public BatchingChannelReader<T> WithTimeout(long millisecondsTimeout)
 	{
-		if (Buffer is null || Buffer.Reader.Completion.IsCompleted) return false;
-		if (TryPipeItems()) return true;
-		if (_batch is null) return false;
+		_timeout = millisecondsTimeout <= 0 ? Timeout.Infinite : millisecondsTimeout;
 
-		lock (Buffer)
+		if (Buffer is null || Buffer.Reader.Completion.IsCompleted)
+			return this;
+
+		if (_timeout == Timeout.Infinite)
 		{
-			if (Buffer.Reader.Completion.IsCompleted) return false;
-			if (TryPipeItems()) return true;
-			List<T>? c = _batch;
-			if (c is null || Buffer.Reader.Completion.IsCompleted)
-				return false;
-			c.TrimExcess();
-			_batch = null;
-			return Buffer.Writer.TryWrite(c); // Should always be true at this point.
+			Interlocked.Exchange(ref _timer, null)?.Dispose();
+			return this;
 		}
+
+		LazyInitializer.EnsureInitialized(ref _timer,
+			() => new Timer(obj => ForceBatch()));
+
+		return this;
 	}
 
+	/// <param name="timeout">
+	/// The timeout value where after a batch is forced.<br/>
+	/// A value of zero or less cancels/clears any timeout.<br/>
+	/// Note: Values are converted to milliseconds.
+	/// </param>
+	/// <inheritdoc cref="WithTimeout(long)"/>
+	public BatchingChannelReader<T> WithTimeout(TimeSpan timeout)
+		=> WithTimeout(TimeSpan.FromMilliseconds(timeout.TotalMilliseconds));
+
 	/// <inheritdoc />
-	protected override bool TryPipeItems()
+	protected override void OnBeforeFinalFlush()
+		=> Interlocked.Exchange(ref _timer, null)?.Dispose();
+
+	/// <inheritdoc />
+	protected override bool TryPipeItems(bool flush)
 	{
 		if (Buffer is null || Buffer.Reader.Completion.IsCompleted)
 			return false;
 
+		var batched = false;
 		lock (Buffer)
 		{
 			if (Buffer.Reader.Completion.IsCompleted) return false;
@@ -65,37 +99,50 @@ protected override bool TryPipeItems()
 			ChannelReader<T>? source = Source;
 			if (source is null || source.Completion.IsCompleted)
 			{
-				// All finished, release the last batch to the buffer.
+				// All finished, if necessary, release the last batch to the buffer.
 				if (c is null) return false;
-
-				c.TrimExcess();
-				_batch = null;
-
-				Buffer.Writer.TryWrite(c);
-				return true;
+				goto flushBatch;
 			}
 
 			while (source.TryRead(out T? item))
 			{
 				if (c is null) _batch = c = new List<T>(_batchSize) { item };
 				else c.Add(item);
 
-				if (c.Count == _batchSize)
-				{
-					_batch = null;
-					Buffer.Writer.TryWrite(c);
-					return true;
-				}
+				Debug.Assert(c.Count <= _batchSize);
+				if (c.Count != _batchSize) continue; // should never be greater.
+
+				_batch = null; // _batch should always have at least 1 item in it.
+				batched = Buffer.Writer.TryWrite(c);
+				Debug.Assert(batched);
+				c = null;
 			}
 
-			return false;
+			if (!flush || c is null)
+				goto finalizeTimer;
+
+			flushBatch:
+
+			c.TrimExcess();
+			_batch = null;
+
+			batched = Buffer.Writer.TryWrite(c);
+			Debug.Assert(batched);
+
+			finalizeTimer:
+
+			var ok = _timer?.Change(_batch is null ? Timeout.Infinite : _timeout, 0);
+			Debug.Assert(ok ?? true);
+
+			return batched;
 		}
 	}
 
 	/// <inheritdoc />
-	protected override async ValueTask<bool> WaitToReadAsyncCore(ValueTask<bool> bufferWait, CancellationToken cancellationToken)
+	protected override async ValueTask<bool> WaitToReadAsyncCore(
+		ValueTask<bool> bufferWait,
+		CancellationToken cancellationToken)
 	{
-
 		ChannelReader<T>? source = Source;
 		if (source is null) return await bufferWait.ConfigureAwait(false);
 
@@ -108,7 +155,7 @@ protected override async ValueTask<bool> WaitToReadAsyncCore(ValueTask<bool> buf
 		if (b.IsCompleted) return await b.ConfigureAwait(false);
 
 		ValueTask<bool> s = source.WaitToReadAsync(token);
-		if (s.IsCompleted && !b.IsCompleted) TryPipeItems();
+		if (s.IsCompleted && !b.IsCompleted) TryPipeItems(false);
 
 		if (b.IsCompleted)
 		{
@@ -123,7 +170,7 @@ protected override async ValueTask<bool> WaitToReadAsyncCore(ValueTask<bool> buf
 			return await b.ConfigureAwait(false);
 		}
 
-		TryPipeItems();
+		TryPipeItems(false);
 		goto start;
 	}
 }

Open.ChannelExtensions/BufferingChannelReader.cs

@@ -44,10 +44,11 @@ protected BufferingChannelReader(ChannelReader<TIn> source, bool singleReader, b
 
 			source.Completion.ContinueWith(t =>
 			{
+				OnBeforeFinalFlush();
 				// Need to be sure writing is done before we continue...
 				lock (Buffer)
 				{
-					while (TryPipeItems()) { }
+					TryPipeItems(true);
 					Buffer.Writer.Complete(t.Exception);
 				}
 
@@ -56,28 +57,38 @@ protected BufferingChannelReader(ChannelReader<TIn> source, bool singleReader, b
 		}
 	}
 
+	/// <summary>
+	/// Called before the last items are flushed to the buffer.
+	/// </summary>
+	protected virtual void OnBeforeFinalFlush()
+	{ }
+
 	private readonly Task _completion;
 	/// <inheritdoc />
 	public override Task Completion => _completion;
 
 	/// <summary>
 	/// The method that triggers adding entries to the buffer.
 	/// </summary>
-	/// <returns></returns>
-	protected abstract bool TryPipeItems();
+	/// <param name="flush">Signals that all items should be piped.</param>
+	/// <returns>True if items were transferred.</returns>
+	protected abstract bool TryPipeItems(bool flush);
 
 	/// <inheritdoc />
 	public override bool TryRead(out TOut item)
 	{
-		if (Buffer is not null) do
+		if (Buffer is not null)
+		{
+			do
 			{
 				if (Buffer.Reader.TryRead(out TOut? i))
 				{
 					item = i;
 					return true;
 				}
 			}
-			while (TryPipeItems());
+			while (TryPipeItems(false));
+		}
 
 		item = default!;
 		return false;
@@ -113,7 +124,7 @@ protected virtual async ValueTask<bool> WaitToReadAsyncCore(ValueTask<bool> buff
 		if (bufferWait.IsCompleted) return await bufferWait.ConfigureAwait(false);
 
 		ValueTask<bool> s = source.WaitToReadAsync(token);
-		if (s.IsCompleted && !bufferWait.IsCompleted) TryPipeItems();
+		if (s.IsCompleted && !bufferWait.IsCompleted) TryPipeItems(false);
 
 		if (bufferWait.IsCompleted)
 		{
@@ -126,7 +137,7 @@ protected virtual async ValueTask<bool> WaitToReadAsyncCore(ValueTask<bool> buff
 			tokenSource.Cancel();
 			return await bufferWait.ConfigureAwait(false);
 		}
-		TryPipeItems();
+		TryPipeItems(false);
 
 		goto start;
 	}

Open.ChannelExtensions/Extensions.Batch.cs

@@ -5,17 +5,22 @@ namespace Open.ChannelExtensions;
 
 public static partial class Extensions
 {
-
-
 	/// <summary>
 	/// Batches results into the batch size provided with a max capacity of batches.
 	/// </summary>
 	/// <typeparam name="T">The output type of the source channel.</typeparam>
 	/// <param name="source">The channel to read from.</param>
-	/// <param name="batchSize">The maximum size of each batch.</param>
+	/// <param name="batchSize">
+	/// The maximum size of each batch.
+	/// Note: setting this value sets the capacity of each batch (reserves memory).
+	/// </param>
 	/// <param name="singleReader">True will cause the resultant reader to optimize for the assumption that no concurrent read operations will occur.</param>
 	/// <param name="allowSynchronousContinuations">True can reduce the amount of scheduling and markedly improve performance, but may produce unexpected or even undesirable behavior.</param>
 	/// <returns>A channel reader containing the batches.</returns>
-	public static BatchingChannelReader<T> Batch<T>(this ChannelReader<T> source, int batchSize, bool singleReader = false, bool allowSynchronousContinuations = false)
+	public static BatchingChannelReader<T> Batch<T>(
+		this ChannelReader<T> source,
+		int batchSize,
+		bool singleReader = false,
+		bool allowSynchronousContinuations = false)
 		=> new(source ?? throw new ArgumentNullException(nameof(source)), batchSize, singleReader, allowSynchronousContinuations);
 }

Open.ChannelExtensions/Extensions.Join.cs

@@ -14,7 +14,7 @@ public JoiningChannelReader(ChannelReader<TList> source, bool singleReader) : ba
 		{
 		}
 
-		protected override bool TryPipeItems()
+		protected override bool TryPipeItems(bool _)
 		{
 			ChannelReader<TList>? source = Source;
 			if (source is null

Open.ChannelExtensions/Extensions.Pipe.cs

@@ -65,9 +65,7 @@ public static ChannelReader<T> PipeTo<T>(this ChannelReader<T> source,
 		if (target is null) throw new ArgumentNullException(nameof(target));
 		Contract.EndContractBlock();
 
-#pragma warning disable CA2012 // Not awaited because it's run by the scheduler and no need to incur the .AsTask() call.
-		_ = PipeTo(source, target.Writer, true, cancellationToken);
-#pragma warning restore CA2012
+		Task.Run(()=>PipeTo(source, target.Writer, true, cancellationToken));
 
 		return target.Reader;
 	}