Final tweaks before release.

Author: electricessence

Open.ChannelExtensions.ComparisonTests/Open.ChannelExtensions.ComparisonTests.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <OutputType>Exe</OutputType>
-    <TargetFramework>net5.0</TargetFramework>
+    <TargetFramework>net6.0</TargetFramework>
   </PropertyGroup>
 
   <ItemGroup>

Open.ChannelExtensions.Tests/BatchTests.cs

@@ -190,4 +190,84 @@ public static async Task ForceBatchTest2()
 			await Task.Delay(500);
 		}));
 	}
+
+	[Fact]
+	public static async Task TimeoutTest0()
+	{
+		var c = Channel.CreateUnbounded<int>(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });
+		BatchingChannelReader<int> reader = c.Reader.Batch(10).WithTimeout(500);
+		var complete = false;
+		_ = Task.Run(async () =>
+		{
+			for (var i = 0; i < 5; i++)
+			{
+				c.Writer.TryWrite(i);
+			}
+
+			await Task.Delay(1000);
+			complete = true;
+			c.Writer.Complete();
+		});
+
+		using var tokenSource = new CancellationTokenSource(6000);
+		Assert.Equal(1, await reader.ReadAllAsync(tokenSource.Token, async (batch, i) =>
+		{
+			switch (i)
+			{
+				case 0:
+					Assert.Equal(5, batch.Count);
+					Assert.False(complete);
+					break;
+
+				default:
+					throw new Exception("Shouldn't arrive here.");
+			}
+			await Task.Delay(100);
+		}));
+	}
+
+
+	[Fact]
+	public static async Task TimeoutTest1()
+	{
+		var c = Channel.CreateUnbounded<int>(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });
+		BatchingChannelReader<int> reader = c.Reader.Batch(10).WithTimeout(500);
+		_ = Task.Run(async () =>
+		{
+			for(var i = 0;i<15;i++)
+			{
+				c.Writer.TryWrite(i);
+			}
+
+			await Task.Delay(1000);
+
+			for (var i = 0; i < 15; i++)
+			{
+				c.Writer.TryWrite(i);
+			}
+			c.Writer.Complete();
+		});
+
+		using var tokenSource = new CancellationTokenSource(6000);
+		Assert.Equal(4, await reader.ReadAllAsync(tokenSource.Token, async (batch, i) =>
+		{
+			switch (i)
+			{
+				case 0:
+				case 2:
+					Assert.Equal(10, batch.Count);
+					break;
+				case 1:
+				case 3:
+					Assert.Equal(5, batch.Count);
+					break;
+
+				default:
+					throw new Exception("Shouldn't arrive here.");
+			}
+			await Task.Delay(100);
+		}));
+	}
+
 }
+

Open.ChannelExtensions/BatchingChannelReader.cs

@@ -90,11 +90,12 @@ 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;
 
+			var batched = false;
+			var newBatch = false;
 			List<T>? c = _batch;
 			ChannelReader<T>? source = Source;
 			if (source is null || source.Completion.IsCompleted)
@@ -106,35 +107,54 @@ protected override bool TryPipeItems(bool flush)
 
 			while (source.TryRead(out T? item))
 			{
-				if (c is null) _batch = c = new List<T>(_batchSize) { item };
-				else c.Add(item);
+				if (c is not null) c.Add(item);
+				else
+				{
+					newBatch = true; // a new batch could start but not be emmited.
+					_batch = c = new List<T>(_batchSize) { item };
+				}
 
 				Debug.Assert(c.Count <= _batchSize);
-				if (c.Count != _batchSize) continue; // should never be greater.
+				var full = c.Count == _batchSize;
+				while (!full && source.TryRead(out item))
+				{
+					c.Add(item);
+					full = c.Count == _batchSize;
+				}
 
-				_batch = null; // _batch should always have at least 1 item in it.
-				batched = Buffer.Writer.TryWrite(c);
-				Debug.Assert(batched);
-				c = null;
+				if (!full) break;
+
+				Emit(ref c);
 			}
 
 			if (!flush || c is null)
 				goto finalizeTimer;
 
-			flushBatch:
+		flushBatch:
 
 			c.TrimExcess();
-			_batch = null;
+			Emit(ref c);
 
-			batched = Buffer.Writer.TryWrite(c);
-			Debug.Assert(batched);
-
-			finalizeTimer:
-
-			var ok = _timer?.Change(_batch is null ? Timeout.Infinite : _timeout, 0);
-			Debug.Assert(ok ?? true);
+		finalizeTimer:
+			
+			// Are we adding to the existing batch (active timeout) or did we create a new one?
+			if (newBatch && _batch is not null)
+			{
+				var ok = _timer?.Change(_timeout, 0);
+				Debug.Assert(ok ?? true);
+			}
 
 			return batched;
+
+			void Emit(ref List<T>? c)
+			{
+				_batch = null;
+				newBatch = false;
+				if (!batched) _timer?.Change(Timeout.Infinite, 0); // Since we're emmitting one, let's ensure the timeout is cancelled.
+				batched = Buffer!.Writer.TryWrite(c!);
+				Debug.Assert(batched);
+				c = null;
+			}
 		}
 	}
 

Open.ChannelExtensions/Documentation.xml

@@ -6,25 +6,27 @@
 		<EnableNETAnalyzers>true</EnableNETAnalyzers>
 		<Nullable>enable</Nullable>
 		<Authors>electricessence</Authors>
-		<Description>A set of extensions for optimizing/simplifying System.Threading.Channels usage.
+		<Description>
+			A set of extensions for optimizing/simplifying System.Threading.Channels usage.
 
-Includes:
-ReadUntilCancelled, ReadAll, ReadAllConcurrently, WriteAll, WriteAllConcurrently, and Pipe operations.
+			Includes:
+			ReadUntilCancelled, ReadAll, ReadAllConcurrently, WriteAll, WriteAllConcurrently, and Pipe operations.
 
-Part of the "Open" set of libraries.</Description>
+			Part of the "Open" set of libraries.
+		</Description>
 		<PackageTags>channel;channels;channel reader;channel writer;threading;tasks;extensions;async</PackageTags>
 		<Copyright>© electricessence (Oren F.) All rights reserved.</Copyright>
 		<PackageProjectUrl>https://github.com/Open-NET-Libraries/Open.ChannelExtensions</PackageProjectUrl>
 		<RepositoryUrl>https://github.com/Open-NET-Libraries/Open.ChannelExtensions</RepositoryUrl>
 		<RepositoryType>git</RepositoryType>
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-		<Version>5.0.0</Version>
-		<PackageReleaseNotes>Updated to System.Threading.Channels and set verson to match.</PackageReleaseNotes>
+		<Version>5.1.0</Version>
+		<PackageReleaseNotes>Added .WithTimeout(milliseconds) method for BatchChannelReader.</PackageReleaseNotes>
 		<PackageLicenseExpression>MIT</PackageLicenseExpression>
 		<PublishRepositoryUrl>true</PublishRepositoryUrl>
 		<IncludeSymbols>true</IncludeSymbols>
 		<SymbolPackageFormat>snupkg</SymbolPackageFormat>
-		<PackageIcon>logo.png</PackageIcon>		
+		<PackageIcon>logo.png</PackageIcon>
 	</PropertyGroup>
 
 	<ItemGroup>