Expanded transaction capability.

Author: electricessence

ExpressiveCommandBase.cs

@@ -15,17 +15,35 @@ public class ExpressiveDbCommand : ExpressiveDbCommandBase<DbConnection, DbComma
         /// <param name="type">The command type>.</param>
         /// <param name="command">The SQL command.</param>
         /// <param name="params">The list of params</param>
-        public ExpressiveDbCommand(IDbConnectionFactory<DbConnection> connFactory, CommandType type, string command, List<Param> @params = null)
-            : base(connFactory, type, command, @params)
+        public ExpressiveDbCommand(IDbConnectionFactory<DbConnection> connFactory, CommandType type, string command, List<Param> @params)
+            : base(connFactory, type, command, @params?.ToList())
         {
         }
 
         /// <param name="connFactory">The factory to generate connections from.</param>
         /// <param name="type">The command type>.</param>
         /// <param name="command">The SQL command.</param>
         /// <param name="params">The list of params</param>
-        protected ExpressiveDbCommand(IDbConnectionFactory<DbConnection> connFactory, CommandType type, string command, params Param[] @params)
-            : this(connFactory, type, command, @params.ToList())
+        public ExpressiveDbCommand(IDbConnectionFactory<DbConnection> connFactory, CommandType type, string command, params Param[] @params)
+            : base(connFactory, type, command, @params?.ToList())
+        {
+        }
+
+        /// <param name="connection">The connection to execute the command on.</param>
+        /// <param name="type">The command type>.</param>
+        /// <param name="command">The SQL command.</param>
+        /// <param name="params">The list of params</param>
+        public ExpressiveDbCommand(DbConnection connection, CommandType type, string command, List<Param> @params)
+            : base(connection, type, command, @params?.ToList())
+        {
+        }
+
+        /// <param name="connection">The connection to execute the command on.</param>
+        /// <param name="type">The command type>.</param>
+        /// <param name="command">The SQL command.</param>
+        /// <param name="params">The list of params</param>
+        public ExpressiveDbCommand(DbConnection connection, CommandType type, string command, params Param[] @params)
+            : this(connection, type, command, @params?.ToList())
         {
         }
 

ExpressiveDbCommand.cs

@@ -28,43 +28,51 @@ public abstract class ExpressiveDbCommandBase<TConnection, TCommand, TDbType, TT
         /// <param name="type">The command type>.</param>
         /// <param name="command">The SQL command.</param>
         /// <param name="params">The list of params</param>
-        public ExpressiveDbCommandBase(
-			IDbConnectionFactory<TConnection> connFactory,
-			CommandType type,
-			string command,
-			List<Param> @params = null)
-			: base(connFactory, type, command, @params)
+        protected ExpressiveDbCommandBase(
+            IDbConnectionFactory<TConnection> connFactory,
+            CommandType type,
+            string command,
+            List<Param> @params)
+            : base(connFactory, type, command, @params)
         {
         }
 
-		/// <param name="connFactory">The factory to generate connections from.</param>
-		/// <param name="type">The command type>.</param>
-		/// <param name="command">The SQL command.</param>
-		/// <param name="params">The list of params</param>
-		protected ExpressiveDbCommandBase(
-			IDbConnectionFactory<TConnection> connFactory,
-			CommandType type,
-			string command,
-			params Param[] @params)
-			: this(connFactory, type, command, @params.ToList())
+        /// <param name="connection">The connection to execute the command on.</param>
+        /// <param name="type">The command type>.</param>
+        /// <param name="command">The SQL command.</param>
+        /// <param name="params">The list of params</param>
+        protected ExpressiveDbCommandBase(
+            TConnection connection,
+            CommandType type,
+            string command,
+            List<Param> @params)
+            : base(connection, type, command, @params)
         {
         }
 
+
         /// <summary>
         /// Asynchronously executes a reader on a command with a handler function.
         /// </summary>
         /// <param name="handler">The handler function for each IDataRecord.</param>
         public async Task ExecuteAsync(Func<TCommand, Task> handler)
         {
-            using (var con = ConnectionFactory.Create())
-            using (var cmd = (TCommand)con.CreateCommand(
+            TConnection con = Connection ?? ConnectionFactory.Create();
+            try
+            {
+                using (var cmd = (TCommand)con.CreateCommand(
                 Type, Command, Timeout))
+                {
+                    var c = cmd as TCommand;
+                    if (c == null) throw new InvalidCastException($"Actual command type ({cmd.GetType()}) is not compatible with expected command type ({typeof(TCommand)}).");
+                    AddParams(c);
+                    await con.OpenAsync();
+                    await handler(c);
+                }
+            }
+            finally
             {
-                var c = cmd as TCommand;
-                if (c == null) throw new InvalidCastException($"Actual command type ({cmd.GetType()}) is not compatible with expected command type ({typeof(TCommand)}).");
-                AddParams(c);
-                await con.OpenAsync();
-                await handler(c);
+                if (Connection == null) con.Dispose();
             }
         }
 
@@ -76,15 +84,21 @@ public async Task ExecuteAsync(Func<TCommand, Task> handler)
         /// <returns>The result of the transform.</returns>
         public async Task<T> ExecuteAsync<T>(Func<TCommand, Task<T>> transform)
         {
-            using (var con = ConnectionFactory.Create())
-            using (var cmd = con.CreateCommand(
-                Type, Command, Timeout))
+            TConnection con = Connection ?? ConnectionFactory.Create();
+            try
             {
-                var c = cmd as TCommand;
-                if (c == null) throw new InvalidCastException($"Actual command type ({cmd.GetType()}) is not compatible with expected command type ({typeof(TCommand)}).");
-                AddParams(c);
-                await con.OpenAsync();
-                return await transform(c);
+                using (var cmd = con.CreateCommand(Type, Command, Timeout))
+                {
+                    var c = cmd as TCommand;
+                    if (c == null) throw new InvalidCastException($"Actual command type ({cmd.GetType()}) is not compatible with expected command type ({typeof(TCommand)}).");
+                    AddParams(c);
+                    await con.OpenAsync();
+                    return await transform(c);
+                }
+            }
+            finally
+            {
+                if (Connection == null) con.Dispose();
             }
         }
 
@@ -94,19 +108,25 @@ public async Task<T> ExecuteAsync<T>(Func<TCommand, Task<T>> transform)
         /// <returns>The value from the return parameter.</returns>
         public async Task<object> ExecuteReturnAsync()
         {
-            using (var con = ConnectionFactory.Create())
-            using (var cmd = con.CreateCommand(
-                Type, Command, Timeout))
+            TConnection con = Connection ?? ConnectionFactory.Create();
+            try
             {
-                var c = cmd as TCommand;
-                if (c == null) throw new InvalidCastException($"Actual command type ({cmd.GetType()}) is not compatible with expected command type ({typeof(TCommand)}).");
-                AddParams(c);
-                var returnParameter = c.CreateParameter();
-                returnParameter.Direction = ParameterDirection.ReturnValue;
-                cmd.Parameters.Add(returnParameter);
-                await con.OpenAsync();
-                await c.ExecuteNonQueryAsync();
-                return returnParameter.Value;
+                using (var cmd = con.CreateCommand(Type, Command, Timeout))
+                {
+                    var c = cmd as TCommand;
+                    if (c == null) throw new InvalidCastException($"Actual command type ({cmd.GetType()}) is not compatible with expected command type ({typeof(TCommand)}).");
+                    AddParams(c);
+                    var returnParameter = c.CreateParameter();
+                    returnParameter.Direction = ParameterDirection.ReturnValue;
+                    cmd.Parameters.Add(returnParameter);
+                    await con.OpenAsync();
+                    await c.ExecuteNonQueryAsync();
+                    return returnParameter.Value;
+                }
+            }
+            finally
+            {
+                if (Connection == null) con.Dispose();
             }
         }
 
@@ -222,29 +242,29 @@ public Task<List<T>> TakeAsync<T>(Func<IDataRecord, T> transform, int count)
         }
 
 
-		/// <summary>
-		/// Reads the first column from every record and returns the results as a list..
-		/// DBNull values are converted to null.
-		/// </summary>
-		/// <returns>The list of transformed records.</returns>
-		public Task<IEnumerable<object>> FirstOrdinalResultsAsync()
-			=> ExecuteAsync(command => command.FirstOrdinalResultsAsync());
-
-		/// <summary>
-		/// Reads the first column from every record..
-		/// DBNull values are converted to null.
-		/// </summary>
-		/// <returns>The enumerable of casted values.</returns>
-		public Task<IEnumerable<T0>> FirstOrdinalResultsAsync<T0>()
-			=> ExecuteAsync(command => command.FirstOrdinalResultsAsync<T0>());
-
-		/// <summary>
-		/// Asynchronously iterates all records within the current result set using an IDataReader and returns the desired results.
-		/// </summary>
-		/// <param name="n">The first ordinal to include in the request to the reader for each record.</param>
-		/// <param name="others">The remaining ordinals to request from the reader for each record.</param>
-		/// <returns>The QueryResult that contains all the results and the column mappings.</returns>
-		public Task<QueryResult<Queue<object[]>>> RetrieveAsync(int n, params int[] others)
+        /// <summary>
+        /// Reads the first column from every record and returns the results as a list..
+        /// DBNull values are converted to null.
+        /// </summary>
+        /// <returns>The list of transformed records.</returns>
+        public Task<IEnumerable<object>> FirstOrdinalResultsAsync()
+            => ExecuteAsync(command => command.FirstOrdinalResultsAsync());
+
+        /// <summary>
+        /// Reads the first column from every record..
+        /// DBNull values are converted to null.
+        /// </summary>
+        /// <returns>The enumerable of casted values.</returns>
+        public Task<IEnumerable<T0>> FirstOrdinalResultsAsync<T0>()
+            => ExecuteAsync(command => command.FirstOrdinalResultsAsync<T0>());
+
+        /// <summary>
+        /// Asynchronously iterates all records within the current result set using an IDataReader and returns the desired results.
+        /// </summary>
+        /// <param name="n">The first ordinal to include in the request to the reader for each record.</param>
+        /// <param name="others">The remaining ordinals to request from the reader for each record.</param>
+        /// <returns>The QueryResult that contains all the results and the column mappings.</returns>
+        public Task<QueryResult<Queue<object[]>>> RetrieveAsync(int n, params int[] others)
             => RetrieveAsync(new int[1] { n }.Concat(others));
 
         /// <summary>

ExpressiveDbCommandBase.cs

@@ -0,0 +1,85 @@
+using System;
+using System.Data;
+using System.Data.Common;
+using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
+
+namespace Open.Database.Extensions
+{
+    public static partial class Extensions
+    {
+        internal static bool IsStillAlive<T>(this ITargetBlock<T> task)
+        {
+            return IsStillAlive(task.Completion);
+        }
+
+        /// <summary>
+        /// Iterates an IDataReader through the transform function and posts each record to the target block.
+        /// </summary>
+        /// <typeparam name="T">The return type of the transform function.</typeparam>
+        /// <param name="reader">The IDataReader to iterate.</param>
+        /// <param name="transform">The transform function for each IDataRecord.</param>
+        /// <param name="target">The target block to receivethe results.</param>
+        public static void ToTargetBlock<T>(this IDataReader reader,
+            ITargetBlock<T> target,
+            Func<IDataRecord, T> transform)
+        {
+            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>
+        public static async Task ToTargetBlockAsync<T>(this DbDataReader reader,
+            ITargetBlock<T> target,
+            Func<IDataRecord, T> transform)
+        {
+            Task<bool> lastSend = null;
+            while (target.IsStillAlive()
+                && await reader.ReadAsync()
+                && (lastSend == null || await lastSend))
+            {
+                // Allows for a premtive read before waiting for the next send.
+                lastSend = target.SendAsync(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="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>
+        public static async Task ToTargetBlockAsync<T>(this DbCommand command,
+            ITargetBlock<T> target,
+            Func<IDataRecord, T> transform)
+        {
+            if (target.IsStillAlive())
+            {
+                using (var reader = await command.ExecuteReaderAsync())
+                {
+                    if (target.IsStillAlive())
+                        await reader.ToTargetBlockAsync(target, transform);
+                }
+            }
+        }
+
+        /// <summary>
+        /// Iterates an IDataReader through the transform function and posts each record to the target block.
+        /// </summary>
+        /// <typeparam name="T">The return type of the transform function.</typeparam>
+        /// <param name="command">The IDataReader to iterate.</param>
+        /// <param name="target">The target block to receive the results.</param>
+        /// <param name="transform">The transform function for each IDataRecord.</param>
+        public static void ToTargetBlock<T>(this IDbCommand command,
+            ITargetBlock<T> target,
+            Func<IDataRecord, T> transform)
+            => command.ExecuteReader(reader => reader.ToTargetBlock(target, transform));
+
+    }
+}