using System; using System.Collections; using System.Collections.Generic; using System.Data; using System.Data.Common; using System.Linq; using System.Threading; using System.Threading.Tasks; namespace Dapper { public static partial class SqlMapper { /// /// Execute a query asynchronously using .NET 4.5 Task. /// public static Task> QueryAsync(this IDbConnection cnn, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null) { return QueryAsync(cnn, typeof(T), new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default(CancellationToken))); } /// /// Execute a query asynchronously using .NET 4.5 Task. /// public static Task> QueryAsync(this IDbConnection cnn, Type type, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null) { if (type == null) throw new ArgumentNullException("type"); return QueryAsync(cnn, type, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default(CancellationToken))); } /// /// Execute a query asynchronously using .NET 4.5 Task. /// public static Task> QueryAsync(this IDbConnection cnn, CommandDefinition command) { return QueryAsync(cnn, typeof(T), command); } /// /// Execute a query asynchronously using .NET 4.5 Task. /// public static Task> QueryAsync(this IDbConnection cnn, Type type, CommandDefinition command) { return QueryAsync(cnn, type, command); } private static async Task> QueryAsync(this IDbConnection cnn, Type effectiveType, CommandDefinition command) { object param = command.Parameters; var identity = new Identity(command.CommandText, command.CommandType, cnn, effectiveType, param == null ? null : param.GetType(), null); var info = GetCacheInfo(identity, param, command.AddToCache); bool wasClosed = cnn.State == ConnectionState.Closed; using (var cmd = (DbCommand)command.SetupCommand(cnn, info.ParamReader)) { try { if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false); using (var reader = await cmd.ExecuteReaderAsync(command.CancellationToken).ConfigureAwait(false)) { return ExecuteReader(reader, effectiveType, identity, info, command.AddToCache).ToList(); } } finally { if (wasClosed) cnn.Close(); } } } /// /// Execute a command asynchronously using .NET 4.5 Task. /// public static Task ExecuteAsync(this IDbConnection cnn, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null) { return ExecuteAsync(cnn, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default(CancellationToken))); } /// /// Execute a command asynchronously using .NET 4.5 Task. /// public static Task ExecuteAsync(this IDbConnection cnn, CommandDefinition command) { object param = command.Parameters; IEnumerable multiExec = param as IEnumerable; if (multiExec != null && !(param is string)) { return ExecuteMultiImplAsync(cnn, command, multiExec); } else { return ExecuteImplAsync(cnn, command, param); } } private struct AsyncExecState { public readonly DbCommand Command; public readonly Task Task; public AsyncExecState(DbCommand command, Task task) { this.Command = command; this.Task = task; } } private static async Task ExecuteMultiImplAsync(IDbConnection cnn, CommandDefinition command, IEnumerable multiExec) { bool isFirst = true; int total = 0; bool wasClosed = cnn.State == ConnectionState.Closed; try { if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false); CacheInfo info = null; string masterSql = null; if ((command.Flags & CommandFlags.Pipelined) != 0) { const int MAX_PENDING = 100; var pending = new Queue(MAX_PENDING); DbCommand cmd = null; try { foreach (var obj in multiExec) { if (isFirst) { isFirst = false; cmd = (DbCommand)command.SetupCommand(cnn, null); masterSql = cmd.CommandText; var identity = new Identity(command.CommandText, cmd.CommandType, cnn, null, obj.GetType(), null); info = GetCacheInfo(identity, obj, command.AddToCache); } else if(pending.Count >= MAX_PENDING) { var recycled = pending.Dequeue(); total += await recycled.Task.ConfigureAwait(false); cmd = recycled.Command; cmd.CommandText = masterSql; // because we do magic replaces on "in" etc cmd.Parameters.Clear(); // current code is Add-tastic } else { cmd = (DbCommand)command.SetupCommand(cnn, null); } info.ParamReader(cmd, obj); var task = cmd.ExecuteNonQueryAsync(command.CancellationToken); pending.Enqueue(new AsyncExecState(cmd, task)); cmd = null; // note the using in the finally: this avoids a double-dispose } while (pending.Count != 0) { var pair = pending.Dequeue(); using (pair.Command) { } // dispose commands total += await pair.Task.ConfigureAwait(false); } } finally { // this only has interesting work to do if there are failures using (cmd) { } // dispose commands while (pending.Count != 0) { // dispose tasks even in failure using (pending.Dequeue().Command) { } // dispose commands } } return total; } else { using (var cmd = (DbCommand)command.SetupCommand(cnn, null)) { foreach (var obj in multiExec) { if (isFirst) { masterSql = cmd.CommandText; isFirst = false; var identity = new Identity(command.CommandText, cmd.CommandType, cnn, null, obj.GetType(), null); info = GetCacheInfo(identity, obj, command.AddToCache); } else { cmd.CommandText = masterSql; // because we do magic replaces on "in" etc cmd.Parameters.Clear(); // current code is Add-tastic } info.ParamReader(cmd, obj); total += await cmd.ExecuteNonQueryAsync(command.CancellationToken).ConfigureAwait(false); } } } } finally { if (wasClosed) cnn.Close(); } return total; } private static async Task ExecuteImplAsync(IDbConnection cnn, CommandDefinition command, object param) { var identity = new Identity(command.CommandText, command.CommandType, cnn, null, param == null ? null : param.GetType(), null); var info = GetCacheInfo(identity, param, command.AddToCache); bool wasClosed = cnn.State == ConnectionState.Closed; using (var cmd = (DbCommand)command.SetupCommand(cnn, info.ParamReader)) { try { if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false); return await cmd.ExecuteNonQueryAsync(command.CancellationToken).ConfigureAwait(false); } finally { if (wasClosed) cnn.Close(); } } } /// /// Maps a query to objects /// /// The first type in the recordset /// The second type in the recordset /// The return type /// /// /// /// /// /// /// The field we should split and read the second object from (default: id) /// Number of seconds before command execution timeout /// Is it a stored proc or a batch? /// public static Task> QueryAsync(this IDbConnection cnn, string sql, Func map, dynamic param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null) { return MultiMapAsync(cnn, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default(CancellationToken)), map, splitOn); } /// /// Maps a query to objects /// /// The first type in the recordset /// The second type in the recordset /// The return type /// /// The field we should split and read the second object from (default: id) /// The command to execute /// /// public static Task> QueryAsync(this IDbConnection cnn, CommandDefinition command, Func map, string splitOn = "Id") { return MultiMapAsync(cnn, command, map, splitOn); } /// /// Maps a query to objects /// /// /// /// /// /// /// /// /// /// /// /// The Field we should split and read the second object from (default: id) /// Number of seconds before command execution timeout /// /// public static Task> QueryAsync(this IDbConnection cnn, string sql, Func map, dynamic param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null) { return MultiMapAsync(cnn, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default(CancellationToken)), map, splitOn); } /// /// Maps a query to objects /// /// /// /// /// /// /// The field we should split and read the second object from (default: id) /// The command to execute /// /// public static Task> QueryAsync(this IDbConnection cnn, CommandDefinition command, Func map, string splitOn = "Id") { return MultiMapAsync(cnn, command, map, splitOn); } /// /// Perform a multi mapping query with 4 input parameters /// /// /// /// /// /// /// /// /// /// /// /// /// /// /// /// public static Task> QueryAsync(this IDbConnection cnn, string sql, Func map, dynamic param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null) { return MultiMapAsync(cnn, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default(CancellationToken)), map, splitOn); } /// /// Perform a multi mapping query with 4 input parameters /// /// /// /// /// /// /// /// The field we should split and read the second object from (default: id) /// The command to execute /// /// public static Task> QueryAsync(this IDbConnection cnn, CommandDefinition command, Func map, string splitOn = "Id") { return MultiMapAsync(cnn, command, map, splitOn); } /// /// Perform a multi mapping query with 5 input parameters /// public static Task> QueryAsync(this IDbConnection cnn, string sql, Func map, dynamic param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null) { return MultiMapAsync(cnn, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default(CancellationToken)), map, splitOn); } /// /// Perform a multi mapping query with 5 input parameters /// public static Task> QueryAsync(this IDbConnection cnn, CommandDefinition command, Func map, string splitOn = "Id") { return MultiMapAsync(cnn, command, map, splitOn); } /// /// Perform a multi mapping query with 6 input parameters /// public static Task> QueryAsync(this IDbConnection cnn, string sql, Func map, dynamic param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null) { return MultiMapAsync(cnn, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default(CancellationToken)), map, splitOn); } /// /// Perform a multi mapping query with 6 input parameters /// public static Task> QueryAsync(this IDbConnection cnn, CommandDefinition command, Func map, string splitOn = "Id") { return MultiMapAsync(cnn, command, map, splitOn); } /// /// Perform a multi mapping query with 7 input parameters /// public static Task> QueryAsync(this IDbConnection cnn, string sql, Func map, dynamic param = null, IDbTransaction transaction = null, bool buffered = true, string splitOn = "Id", int? commandTimeout = null, CommandType? commandType = null) { return MultiMapAsync(cnn, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, buffered ? CommandFlags.Buffered : CommandFlags.None, default(CancellationToken)), map, splitOn); } /// /// Perform a multi mapping query with 7 input parameters /// public static Task> QueryAsync(this IDbConnection cnn, CommandDefinition command, Func map, string splitOn = "Id") { return MultiMapAsync(cnn, command, map, splitOn); } private static async Task> MultiMapAsync(this IDbConnection cnn, CommandDefinition command, Delegate map, string splitOn) { object param = command.Parameters; var identity = new Identity(command.CommandText, command.CommandType, cnn, typeof(TFirst), param == null ? null : param.GetType(), new[] { typeof(TFirst), typeof(TSecond), typeof(TThird), typeof(TFourth), typeof(TFifth), typeof(TSixth), typeof(TSeventh) }); var info = GetCacheInfo(identity, param, command.AddToCache); bool wasClosed = cnn.State == ConnectionState.Closed; try { if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false); using (var cmd = (DbCommand)command.SetupCommand(cnn, info.ParamReader)) using (var reader = await cmd.ExecuteReaderAsync(command.CancellationToken).ConfigureAwait(false)) { var results = MultiMapImpl(null, default(CommandDefinition), map, splitOn, reader, identity); return command.Buffered ? results.ToList() : results; } } finally { if (wasClosed) cnn.Close(); } } private static IEnumerable ExecuteReader(IDataReader reader, Type effectiveType, Identity identity, CacheInfo info, bool addToCache) { var tuple = info.Deserializer; int hash = GetColumnHash(reader); if (tuple.Func == null || tuple.Hash != hash) { tuple = info.Deserializer = new DeserializerState(hash, GetDeserializer(effectiveType, reader, 0, -1, false)); if(addToCache) SetQueryCache(identity, info); } var func = tuple.Func; while (reader.Read()) { yield return (T)func(reader); } } /// /// Execute a command that returns multiple result sets, and access each in turn /// public static Task QueryMultipleAsync( #if CSHARP30 this IDbConnection cnn, string sql, object param, IDbTransaction transaction, int? commandTimeout, CommandType? commandType #else this IDbConnection cnn, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null #endif ) { var command = new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered); return QueryMultipleAsync(cnn, command); } /// /// Execute a command that returns multiple result sets, and access each in turn /// public static async Task QueryMultipleAsync(this IDbConnection cnn, CommandDefinition command) { object param = command.Parameters; Identity identity = new Identity(command.CommandText, command.CommandType, cnn, typeof(GridReader), param == null ? null : param.GetType(), null); CacheInfo info = GetCacheInfo(identity, param, command.AddToCache); DbCommand cmd = null; IDataReader reader = null; bool wasClosed = cnn.State == ConnectionState.Closed; try { if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false); cmd = (DbCommand)command.SetupCommand(cnn, info.ParamReader); reader = await cmd.ExecuteReaderAsync(wasClosed ? CommandBehavior.CloseConnection : CommandBehavior.Default, command.CancellationToken).ConfigureAwait(false); var result = new GridReader(cmd, reader, identity); wasClosed = false; // *if* the connection was closed and we got this far, then we now have a reader // with the CloseConnection flag, so the reader will deal with the connection; we // still need something in the "finally" to ensure that broken SQL still results // in the connection closing itself return result; } catch { if (reader != null) { if (!reader.IsClosed) try { cmd.Cancel(); } catch { /* don't spoil the existing exception */ } reader.Dispose(); } if (cmd != null) cmd.Dispose(); if (wasClosed) cnn.Close(); throw; } } /// /// Execute parameterized SQL and return an /// /// An that can be used to iterate over the results of the SQL query. /// /// This is typically used when the results of a query are not processed by Dapper, for example, used to fill a /// or . /// /// /// /// /// /// public static Task ExecuteReaderAsync( #if CSHARP30 this IDbConnection cnn, string sql, object param, IDbTransaction transaction, int? commandTimeout, CommandType? commandType #else this IDbConnection cnn, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null #endif ) { var command = new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered); return ExecuteReaderImplAsync(cnn, command); } /// /// Execute parameterized SQL and return an /// /// An that can be used to iterate over the results of the SQL query. /// /// This is typically used when the results of a query are not processed by Dapper, for example, used to fill a /// or . /// public static Task ExecuteReaderAsync(this IDbConnection cnn, CommandDefinition command) { return ExecuteReaderImplAsync(cnn, command); } private static async Task ExecuteReaderImplAsync(IDbConnection cnn, CommandDefinition command) { Action paramReader = GetParameterReader(cnn, ref command); DbCommand cmd = null; bool wasClosed = cnn.State == ConnectionState.Closed; try { cmd = (DbCommand)command.SetupCommand(cnn, paramReader); if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false); var reader = await cmd.ExecuteReaderAsync(wasClosed ? CommandBehavior.CloseConnection : CommandBehavior.Default, command.CancellationToken).ConfigureAwait(false); wasClosed = false; return reader; } finally { if (wasClosed) cnn.Close(); if (cmd != null) cmd.Dispose(); } } /// /// Execute parameterized SQL that selects a single value /// /// The first cell selected public static Task ExecuteScalarAsync( #if CSHARP30 this IDbConnection cnn, string sql, object param, IDbTransaction transaction, int? commandTimeout, CommandType? commandType #else this IDbConnection cnn, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null #endif ) { var command = new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered); return ExecuteScalarImplAsync(cnn, command); } /// /// Execute parameterized SQL that selects a single value /// /// The first cell selected public static Task ExecuteScalarAsync( #if CSHARP30 this IDbConnection cnn, string sql, object param, IDbTransaction transaction, int? commandTimeout, CommandType? commandType #else this IDbConnection cnn, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null #endif ) { var command = new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered); return ExecuteScalarImplAsync(cnn, command); } /// /// Execute parameterized SQL that selects a single value /// /// The first cell selected public static Task ExecuteScalarAsync(this IDbConnection cnn, CommandDefinition command) { return ExecuteScalarImplAsync(cnn, command); } /// /// Execute parameterized SQL that selects a single value /// /// The first cell selected public static Task ExecuteScalarAsync(this IDbConnection cnn, CommandDefinition command) { return ExecuteScalarImplAsync(cnn, command); } private async static Task ExecuteScalarImplAsync(IDbConnection cnn, CommandDefinition command) { Action paramReader = null; object param = command.Parameters; if (param != null) { var identity = new Identity(command.CommandText, command.CommandType, cnn, null, param.GetType(), null); paramReader = GetCacheInfo(identity, command.Parameters, command.AddToCache).ParamReader; } DbCommand cmd = null; bool wasClosed = cnn.State == ConnectionState.Closed; object result; try { cmd = (DbCommand)command.SetupCommand(cnn, paramReader); if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false); result = await cmd.ExecuteScalarAsync(command.CancellationToken).ConfigureAwait(false); } finally { if (wasClosed) cnn.Close(); if (cmd != null) cmd.Dispose(); } return Parse(result); } } }