Improved Dataflow behavior to include faulting and cancellation.

Author: Oren (electricessence)

Documentation.xml

@@ -4,6 +4,7 @@
 using System.Data.Common;
 using System.Diagnostics.Contracts;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;
 // ReSharper disable MemberCanBePrivate.Global
@@ -429,8 +430,8 @@ public IReceivableSourceBlock<T> AsSourceBlockAsync<T>(
 				{
 					if (t.IsFaulted) ((ITargetBlock<T>)source).Fault(t.Exception);
 					else source.Complete();
-				}, CancellationToken)
-				.ConfigureAwait(false);
+				}, CancellationToken);
+
 			return source;
 		}
 
@@ -501,14 +502,14 @@ public async Task<List<T>> ToListAsync<T>(Func<IDataRecord, T> transform, Comman
 		public Task<IEnumerable<T>> ResultsAsync<T>(params (string Field, string Column)[] fieldMappingOverrides) where T : new()
 			=> ResultsAsync<T>(fieldMappingOverrides as IEnumerable<(string Field, string Column)>);
 
-		/// <summary>
-		/// Returns a source block as the source of records.
-		/// </summary>
-		/// <typeparam name="T">The model type to map the values to (using reflection).</typeparam>
-		/// <param name="fieldMappingOverrides">An override map of field names to column names where the keys are the property names, and values are the column names.</param>
-		/// <param name="options"></param>
-		/// <returns>A transform block that is receiving the results.</returns>
-		public IReceivableSourceBlock<T> AsSourceBlockAsync<T>(
+        /// <summary>
+        /// Returns a source block as the source of records.
+        /// </summary>
+        /// <typeparam name="T">The model type to map the values to (using reflection).</typeparam>
+        /// <param name="fieldMappingOverrides">An override map of field names to column names where the keys are the property names, and values are the column names.</param>
+		/// <param name="options">The optional ExecutionDataflowBlockOptions to use with the source.</param>
+        /// <returns>A transform block that is receiving the results.</returns>
+        public IReceivableSourceBlock<T> AsSourceBlockAsync<T>(
 			IEnumerable<(string Field, string Column)> fieldMappingOverrides,
 			ExecutionDataflowBlockOptions options = null)
 			where T : new()
@@ -517,22 +518,27 @@ public IReceivableSourceBlock<T> AsSourceBlockAsync<T>(
 			var cn = x.ColumnNames;
 			var block = x.ResultsBlock(out var initColumnNames, options);
 
-			ExecuteReaderAsync(async reader =>
-			{
-				// Ignores fields that don't match.
-				var columns = reader.GetMatchingOrdinals(cn, true);
-
-				var ordinalValues = columns.Select(c => c.Ordinal).ToArray();
-				initColumnNames(columns.Select(c => c.Name).ToArray());
-
-				await reader.ToTargetBlockAsync(block,
-					r => r.GetValuesFromOrdinals(ordinalValues),
-					UseAsyncRead);
-
-				block.Complete();
-			});
-
-			return block;
+			ExecuteReaderAsync(
+                reader =>
+			    {
+				    // Ignores fields that don't match.
+				    var columns = reader.GetMatchingOrdinals(cn, true);
+
+				    var ordinalValues = columns.Select(c => c.Ordinal).ToArray();
+				    initColumnNames(columns.Select(c => c.Name).ToArray());
+
+				    return reader.ToTargetBlockAsync(block,
+					    r => r.GetValuesFromOrdinals(ordinalValues),
+					    UseAsyncRead,
+                        CancellationToken);
+			    })
+                .ContinueWith(t =>
+                {
+                    if (t.IsFaulted) ((ITargetBlock<object[]>)block).Fault(t.Exception);
+                    else block.Complete();
+                }, CancellationToken);
+
+            return block;
 		}
 
 		/// <summary>

ExpressiveDbCommandBase.cs

@@ -2,6 +2,7 @@
 using System.Data;
 using System.Data.Common;
 using System.Diagnostics.Contracts;
+using System.Threading;
 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;
 // ReSharper disable MemberCanBePrivate.Global
@@ -34,18 +35,20 @@ public static void ToTargetBlock<T>(this IDataReader reader,
 			while (target.IsStillAlive() && reader.Read() && target.Post(transform(reader))) { }
 		}
 
-		/// <summary>
-		/// Asynchronously iterates an IDataReader and through the transform function and posts each record it to the target block.
-		/// </summary>
-		/// <typeparam name="T">The return type of the transform function.</typeparam>
-		/// <param name="reader">The SqlDataReader to read from.</param>
-		/// <param name="target">The target block to receive the results.</param>
-		/// <param name="transform">The transform function to process each IDataRecord.</param>
-		/// <param name="useReadAsync">If true (default) will iterate the results using .ReadAsync() otherwise will only Execute the reader asynchronously and then use .Read() to iterate the results but still allowing cancellation.</param>
-		public static async Task ToTargetBlockAsync<T>(this DbDataReader reader,
+        /// <summary>
+        /// Asynchronously iterates an IDataReader and through the transform function and posts each record it to the target block.
+        /// </summary>
+        /// <typeparam name="T">The return type of the transform function.</typeparam>
+        /// <param name="reader">The SqlDataReader to read from.</param>
+        /// <param name="target">The target block to receive the results.</param>
+        /// <param name="transform">The transform function to process each IDataRecord.</param>
+        /// <param name="useReadAsync">If true (default) will iterate the results using .ReadAsync() otherwise will only Execute the reader asynchronously and then use .Read() to iterate the results but still allowing cancellation.</param>
+        /// <param name="cancellationToken">Optional cancellation token.</param>
+        public static async Task ToTargetBlockAsync<T>(this DbDataReader reader,
 			ITargetBlock<T> target,
 			Func<IDataRecord, T> transform,
-			bool useReadAsync = true)
+			bool useReadAsync = true,
+            CancellationToken cancellationToken = default)
 		{
 			if (target == null) throw new ArgumentNullException(nameof(target));
 			if (transform == null) throw new ArgumentNullException(nameof(transform));
@@ -54,53 +57,61 @@ public static async Task ToTargetBlockAsync<T>(this DbDataReader reader,
 			if (useReadAsync)
 			{
 				Task<bool> lastSend = null;
-				while (target.IsStillAlive()
-					&& await reader.ReadAsync().ConfigureAwait(false)
+				while (
+                    target.IsStillAlive() && !cancellationToken.IsCancellationRequested
+                    && await reader.ReadAsync(cancellationToken).ConfigureAwait(false) // Premtively grab next while waiting for previous transform.
 					&& (lastSend == null || await lastSend.ConfigureAwait(false)))
 				{
 					var values = transform(reader);
 					lastSend = target.Post(values) ? null : target.SendAsync(values);
 				}
+                // Makes sure we hook up to the last one if the while loop is done to cover any edge cases.
+                if (lastSend != null)
+                    await lastSend.ConfigureAwait(false);
 			}
 			else
 			{
 				var ok = true;
-				while (ok && target.IsStillAlive() && reader.Read())
+				while (ok
+                    && target.IsStillAlive() && !cancellationToken.IsCancellationRequested
+                    && reader.Read())
 				{
 					var values = transform(reader);
 					ok = target.Post(values) || await target.SendAsync(values);
 				}
 			}
 		}
 
-		/// <summary>
-		/// Asynchronously iterates an IDataReader and through the transform function and posts each record it to the target block.
-		/// If a connection is desired to remain open after completion, you must open the connection before calling this method.
-		/// </summary>
-		/// <typeparam name="T">The return type of the transform function.</typeparam>
-		/// <param name="command">The DbCommand to generate a reader from.</param>
-		/// <param name="target">The target block to receive the results.</param>
-		/// <param name="transform">The transform function for each IDataRecord.</param>
-		/// <param name="behavior">The behavior to use with the data reader.</param>
-		/// <param name="useReadAsync">If true (default) will iterate the results using .ReadAsync() otherwise will only Execute the reader asynchronously and then use .Read() to iterate the results but still allowing cancellation.</param>
-		public static async Task ToTargetBlockAsync<T>(this DbCommand command,
+        /// <summary>
+        /// Asynchronously iterates an IDataReader and through the transform function and posts each record it to the target block.
+        /// If a connection is desired to remain open after completion, you must open the connection before calling this method.
+        /// </summary>
+        /// <typeparam name="T">The return type of the transform function.</typeparam>
+        /// <param name="command">The DbCommand to generate a reader from.</param>
+        /// <param name="target">The target block to receive the results.</param>
+        /// <param name="transform">The transform function for each IDataRecord.</param>
+        /// <param name="behavior">The behavior to use with the data reader.</param>
+        /// <param name="useReadAsync">If true (default) will iterate the results using .ReadAsync() otherwise will only Execute the reader asynchronously and then use .Read() to iterate the results but still allowing cancellation.</param>
+        /// <param name="cancellationToken"></param>
+        public static async Task ToTargetBlockAsync<T>(this DbCommand command,
 			ITargetBlock<T> target,
 			Func<IDataRecord, T> transform,
 			CommandBehavior behavior = CommandBehavior.Default,
-			bool useReadAsync = true)
+			bool useReadAsync = true,
+            CancellationToken cancellationToken = default)
 		{
 			if (target == null) throw new ArgumentNullException(nameof(target));
 			if (transform == null) throw new ArgumentNullException(nameof(transform));
 			Contract.EndContractBlock();
 
 			if (target.IsStillAlive())
 			{
-				var state = await command.Connection.EnsureOpenAsync();
+				var state = await command.Connection.EnsureOpenAsync(cancellationToken);
 				if (state == ConnectionState.Closed) behavior = behavior | CommandBehavior.CloseConnection;
-				using (var reader = await command.ExecuteReaderAsync(behavior))
+				using (var reader = await command.ExecuteReaderAsync(behavior, cancellationToken))
 				{
 					if (target.IsStillAlive())
-						await reader.ToTargetBlockAsync(target, transform, useReadAsync);
+						await reader.ToTargetBlockAsync(target, transform, useReadAsync, cancellationToken);
 				}
 			}
 		}

Extensions.Dataflow.cs

@@ -11,11 +11,10 @@
     <Description>Useful set of utilities and abstractions for simplifying modern data-access operations and ensuring DI compatibility.</Description>
     <RepositoryUrl>https://github.com/electricessence/Open.Database.Extensions</RepositoryUrl>
     <RepositoryType>git</RepositoryType>
-    <Version>5.13.2</Version>
-    <AssemblyVersion>5.13.0.0</AssemblyVersion>
-    <FileVersion>5.13.0.0</FileVersion>
-    <PackageReleaseNotes>Cleanup and inspection fixes with added Dataflow options.
-Improved Expressive Command DB-Open behavior to occur before command is created to ensure compatibility with other DbCommands like Snowflake.</PackageReleaseNotes>
+    <Version>5.13.3</Version>
+    <AssemblyVersion>5.13.3.0</AssemblyVersion>
+    <FileVersion>5.13.3.0</FileVersion>
+    <PackageReleaseNotes>Improved Dataflow behavior to include faulting and cancellation.</PackageReleaseNotes>
     <LangVersion>latest</LangVersion>
   </PropertyGroup>