Extended cancellation capability.

Author: Oren (electricessence)

Extensions._.cs

@@ -54,6 +54,7 @@ public static ISourceBlock<T[]> Batch<T>(this ISourceBlock<T> source,
             var batchBlock = dataflowBlockOptions == null
                 ? new BatchBlock<T>(batchSize)
                 : new BatchBlock<T>(batchSize, dataflowBlockOptions);
+
             source.LinkToWithCompletion(batchBlock);
             return batchBlock;
         }
@@ -99,16 +100,20 @@ async Task<int> ToTargetBlockAsyncCore()
                 foreach (var entry in source)
                 {
                     if (cancellationToken.IsCancellationRequested
-                        || !target.Post(entry) && !await target.SendAsync(entry))
+                        || !target.Post(entry) && !await target.SendAsync(entry, cancellationToken))
                         break;
 
                     count++;
                 }
+
+                cancellationToken.ThrowIfCancellationRequested();
+
                 return count;
             }
         }
 
-        public static async Task<List<T>> ToListAsync<T>(this IReceivableSourceBlock<T> source)
+        public static async Task<List<T>> ToListAsync<T>(this IReceivableSourceBlock<T> source,
+            CancellationToken cancellationToken = default)
         {
             if (source == null)
                 throw new NullReferenceException();           
@@ -117,12 +122,16 @@ public static async Task<List<T>> ToListAsync<T>(this IReceivableSourceBlock<T>
             var result = new List<T>();
             do
             {
-                while (source.TryReceive(null, out var e))
+                while (!cancellationToken.IsCancellationRequested
+                    && source.TryReceive(null, out var e))
                 {
                     result.Add(e);
                 }
             }
-            while (await source.OutputAvailableAsync());
+            while (!cancellationToken.IsCancellationRequested
+                && await source.OutputAvailableAsync(cancellationToken));
+
+            cancellationToken.ThrowIfCancellationRequested();
 
             return result;
         }
@@ -163,23 +172,27 @@ async Task CallCompleteWhenFinished()
 
         public static async Task<int> AllLinesTo(this TextReader source,
             ITargetBlock<string> target,
-            bool completeAndWait = false)
+            bool completeAndAwait = false,
+            CancellationToken cancellationToken = default)
         {
             if (source == null)
                 throw new NullReferenceException();
             Contract.EndContractBlock();
 
             var count = 0;
             string line;
-            while ((line = await source.ReadLineAsync()) != null)
+            while (!cancellationToken.IsCancellationRequested
+                && (line = await source.ReadLineAsync()) != null)
             {
-                if (!target.Post(line) && !await target.SendAsync(line))
+                if (!target.Post(line) && !await target.SendAsync(line, cancellationToken))
                     break;
 
                 count++;
             }
 
-            if (completeAndWait)
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (completeAndAwait)
                 await target.CompleteAsync();
 
             return count;
@@ -188,24 +201,28 @@ public static async Task<int> AllLinesTo(this TextReader source,
         public static async Task<int> AllLinesTo<T>(this TextReader source,
             ITargetBlock<T> target,
             Func<string, T> transform,
-            bool completeAndWait = false)
+            bool completeAndAwait = false,
+            CancellationToken cancellationToken = default)
         {
             if (source == null)
                 throw new NullReferenceException();
             Contract.EndContractBlock();
 
             var count = 0;
             string line;
-            while ((line = await source.ReadLineAsync()) != null)
+            while (!cancellationToken.IsCancellationRequested
+                && (line = await source.ReadLineAsync()) != null)
             {
                 var e = transform(line);
-                if (!target.Post(e) && !await target.SendAsync(e))
+                if (!target.Post(e) && !await target.SendAsync(e, cancellationToken))
                     break;
 
                 count++;
             }
 
-            if (completeAndWait)
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (completeAndAwait)
                 await target.CompleteAsync();
 
             return count;

Open.Threading.Dataflow.csproj

@@ -15,10 +15,10 @@ Part of the "Open" set of libraries.</Description>
     <RepositoryUrl>https://github.com/electricessence/Open.Threading.Dataflow/</RepositoryUrl>
     <RepositoryType>git</RepositoryType>
     <PackageTags>dotnet, dotnet-core, dotnetcore, cs, dataflow, tpl, extensions</PackageTags>
-    <Version>2.1.1</Version>
+    <Version>2.1.2</Version>
     <PackageReleaseNotes></PackageReleaseNotes>
-    <AssemblyVersion>2.1.1.0</AssemblyVersion>
-    <FileVersion>2.1.1.0</FileVersion>
+    <AssemblyVersion>2.1.2.0</AssemblyVersion>
+    <FileVersion>2.1.2.0</FileVersion>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
   </PropertyGroup>