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, new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default(CancellationToken)));
}
///
/// Execute a query asynchronously using .NET 4.5 Task.
///
public static async Task> QueryAsync(this IDbConnection cnn, CommandDefinition command)
{
object param = command.Parameters;
var identity = new Identity(command.CommandText, command.CommandType, cnn, typeof(T), param == null ? null : param.GetType(), null);
var info = GetCacheInfo(identity, param);
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, identity, info).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);
} 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);
}
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);
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);
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, Identity identity, CacheInfo info)
{
var tuple = info.Deserializer;
int hash = GetColumnHash(reader);
if (tuple.Func == null || tuple.Hash != hash)
{
tuple = info.Deserializer = new DeserializerState(hash, GetDeserializer(typeof(T), reader, 0, -1, false));
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);
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();
}
}
}
}