Fixed batching.

Author: Oren (electricessence)

Open.ChannelExtensions/BufferingChannelReader.cs

@@ -18,15 +18,17 @@ public BufferingChannelReader(ChannelReader<TIn> source, bool singleReader)
 			Buffer = Extensions.CreateChannel<TOut>(-1, singleReader);
 
 			if (Source.Completion.IsCompleted)
+			{
 				Buffer.Writer.Complete(Source.Completion.Exception);
+			}
 			else
 			{
 				Source.Completion.ContinueWith(t =>
 				{
 					// Need to be sure writing is done before we continue...
 					lock (Buffer)
 					{
-						TryPipeItems();
+						while (TryPipeItems()) { }
 						Buffer.Writer.Complete(t.Exception);
 					}
 				});

Open.ChannelExtensions/Extensions.Batch.cs

@@ -18,7 +18,7 @@ public BatchingChannelReader(ChannelReader<T> source, int batchSize, bool single
 				Contract.EndContractBlock();
 
 				_batchSize = batchSize;
-				_current = new List<T>(batchSize);
+				_current = source.Completion.IsCompleted ? null : new List<T>(batchSize);
 			}
 
 
@@ -36,6 +36,9 @@ protected override bool TryPipeItems()
 					if (c == null)
 						return false;
 
+					if (Buffer.Reader.Completion.IsCompleted)
+						return false;
+
 					if (Source.Completion.IsCompleted)
 					{
 						c.TrimExcess();
@@ -47,22 +50,19 @@ protected override bool TryPipeItems()
 						return true;
 					}
 
-					bool piped = false;
 					while (Source.TryRead(out T item))
 					{
-						piped = true;
-
 						if (c.Count == _batchSize)
 						{
-							_current = new List<T>(_batchSize);
+							_current = new List<T>(_batchSize) { item };
 							Buffer.Writer.TryWrite(c);
-							break;
+							return true;
 						}
 
 						c.Add(item);
 					}
 
-					return piped;
+					return false;
 				}
 			}
 		}