1.30: better async cancellation

Dapper NET40/Properties/AssemblyInfo.cs

@@ -32,5 +32,5 @@
 //
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("1.29.0.0")]
-[assembly: AssemblyFileVersion("1.29.0.0")]
+[assembly: AssemblyVersion("1.30.0.0")]
+[assembly: AssemblyFileVersion("1.30.0.0")]

Dapper NET40/SqlMapper.cs

@@ -1231,7 +1231,7 @@ private static int ExecuteImpl(this IDbConnection cnn, ref CommandDefinition com
 )
         {
             var command = new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered);
-            return ExecuteReaderImpl(cnn, ref command);
+            return ExecuteReaderImpl(cnn, ref command, CommandBehavior.Default);
         }
 
         /// <summary>
@@ -1244,7 +1244,19 @@ private static int ExecuteImpl(this IDbConnection cnn, ref CommandDefinition com
         /// </remarks>
         public static IDataReader ExecuteReader(this IDbConnection cnn, CommandDefinition command)
         {
-            return ExecuteReaderImpl(cnn, ref command);
+            return ExecuteReaderImpl(cnn, ref command, CommandBehavior.Default);
+        }
+        /// <summary>
+        /// Execute parameterized SQL and return an <see cref="IDataReader"/>
+        /// </summary>
+        /// <returns>An <see cref="IDataReader"/> that can be used to iterate over the results of the SQL query.</returns>
+        /// <remarks>
+        /// This is typically used when the results of a query are not processed by Dapper, for example, used to fill a <see cref="DataTable"/>
+        /// or <see cref="DataSet"/>.
+        /// </remarks>
+        public static IDataReader ExecuteReader(this IDbConnection cnn, CommandDefinition command, CommandBehavior commandBehavior)
+        {
+            return ExecuteReaderImpl(cnn, ref command, commandBehavior);
         }
 
 #if !CSHARP30
@@ -1385,7 +1397,7 @@ private static GridReader QueryMultipleImpl(this IDbConnection cnn, ref CommandD
             {
                 if (wasClosed) cnn.Open();
                 cmd = command.SetupCommand(cnn, info.ParamReader);
-                reader = cmd.ExecuteReader(wasClosed ? CommandBehavior.CloseConnection : CommandBehavior.Default);
+                reader = cmd.ExecuteReader(wasClosed ? CommandBehavior.CloseConnection | CommandBehavior.SequentialAccess : CommandBehavior.SequentialAccess);
 
                 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
@@ -1423,7 +1435,7 @@ private static IEnumerable<T> QueryImpl<T>(this IDbConnection cnn, CommandDefini
                 cmd = command.SetupCommand(cnn, info.ParamReader);
 
                 if (wasClosed) cnn.Open();
-                reader = cmd.ExecuteReader(wasClosed ? CommandBehavior.CloseConnection : CommandBehavior.Default);
+                reader = cmd.ExecuteReader(wasClosed ? CommandBehavior.CloseConnection | CommandBehavior.SequentialAccess : CommandBehavior.SequentialAccess);
                 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
@@ -1657,7 +1669,7 @@ partial class DontMap { }
                 {
                     ownedCommand = command.SetupCommand(cnn, cinfo.ParamReader);
                     if (wasClosed) cnn.Open();
-                    ownedReader = ownedCommand.ExecuteReader();
+                    ownedReader = ownedCommand.ExecuteReader(wasClosed ? CommandBehavior.CloseConnection | CommandBehavior.SequentialAccess : CommandBehavior.SequentialAccess);
                     reader = ownedReader;
                 }
                 DeserializerState deserializer = default(DeserializerState);
@@ -3008,7 +3020,7 @@ private static T ExecuteScalarImpl<T>(IDbConnection cnn, ref CommandDefinition c
             return Parse<T>(result);
         }
 
-        private static IDataReader ExecuteReaderImpl(IDbConnection cnn, ref CommandDefinition command)
+        private static IDataReader ExecuteReaderImpl(IDbConnection cnn, ref CommandDefinition command, CommandBehavior commandBehavior)
         {
             Action<IDbCommand, object> paramReader = GetParameterReader(cnn, ref command);
 
@@ -3018,8 +3030,9 @@ private static IDataReader ExecuteReaderImpl(IDbConnection cnn, ref CommandDefin
             {
                 cmd = command.SetupCommand(cnn, paramReader);
                 if (wasClosed) cnn.Open();
-                var reader = cmd.ExecuteReader(wasClosed ? CommandBehavior.CloseConnection : CommandBehavior.Default);
-                wasClosed = false;
+                if (wasClosed) commandBehavior |= CommandBehavior.CloseConnection;
+                var reader = cmd.ExecuteReader(commandBehavior);
+                wasClosed = false; // don't dispose before giving it to them!
                 return reader;
             }
             finally
@@ -3733,7 +3746,7 @@ internal GridReader(IDbCommand command, IDataReader reader, Identity identity)
             /// </summary>
             public IEnumerable<dynamic> Read(bool buffered = true)
             {
-                return Read<DapperRow>(buffered);
+                return ReadImpl<dynamic>(typeof(DapperRow), buffered);
             }
 #endif
 
@@ -3755,21 +3768,7 @@ public IEnumerable<T> Read<T>(bool buffered)
             public IEnumerable<T> Read<T>(bool buffered = true)
 #endif
             {
-                if (reader == null) throw new ObjectDisposedException(GetType().FullName, "The reader has been disposed; this can happen after all data has been consumed");
-                if (consumed) throw new InvalidOperationException("Query results must be consumed in the correct order, and each result can only be consumed once");
-                var typedIdentity = identity.ForGrid(typeof(T), gridIndex);
-                CacheInfo cache = GetCacheInfo(typedIdentity, null, true);
-                var deserializer = cache.Deserializer;
-
-                int hash = GetColumnHash(reader);
-                if (deserializer.Func == null || deserializer.Hash != hash)
-                {
-                    deserializer = new DeserializerState(hash, GetDeserializer(typeof(T), reader, 0, -1, false));
-                    cache.Deserializer = deserializer;
-                }
-                consumed = true;
-                var result = ReadDeferred<T>(gridIndex, deserializer.Func, typedIdentity);
-                return buffered ? result.ToList() : result;
+                return ReadImpl<T>(typeof(T), buffered);
             }
 
             /// <summary>
@@ -3782,6 +3781,11 @@ public IEnumerable<object> Read(Type type, bool buffered = true)
 #endif
             {
                 if (type == null) throw new ArgumentNullException("type");
+                return ReadImpl<object>(type, buffered);
+            }
+
+            private IEnumerable<T> ReadImpl<T>(Type type, bool buffered)
+            {
                 if (reader == null) throw new ObjectDisposedException(GetType().FullName, "The reader has been disposed; this can happen after all data has been consumed");
                 if (consumed) throw new InvalidOperationException("Query results must be consumed in the correct order, and each result can only be consumed once");
                 var typedIdentity = identity.ForGrid(type, gridIndex);
@@ -3795,10 +3799,11 @@ public IEnumerable<object> Read(Type type, bool buffered = true)
                     cache.Deserializer = deserializer;
                 }
                 consumed = true;
-                var result = ReadDeferred<object>(gridIndex, deserializer.Func, typedIdentity);
+                var result = ReadDeferred<T>(gridIndex, deserializer.Func, typedIdentity);
                 return buffered ? result.ToList() : result;
             }
 
+
             private IEnumerable<TReturn> MultiReadInternal<TFirst, TSecond, TThird, TFourth, TFifth, TSixth, TSeventh, TReturn>(Delegate func, string splitOn)
             {
                 var identity = this.identity.ForGrid(typeof(TReturn), new Type[] { 
@@ -3965,7 +3970,6 @@ private void NextResult()
 
                     Dispose();
                 }
-
             }
             /// <summary>
             /// Dispose the grid, closing and disposing both the underlying reader and command.

Dapper NET45/SqlMapperAsync.cs

@@ -13,6 +13,23 @@ namespace Dapper
 
     public static partial class SqlMapper
     {
+
+        /// <summary>
+        /// Execute a query asynchronously using .NET 4.5 Task.
+        /// </summary>
+        public static Task<IEnumerable<dynamic>> QueryAsync(this IDbConnection cnn, string sql, dynamic param = null, IDbTransaction transaction = null, int? commandTimeout = null, CommandType? commandType = null)
+        {
+            return QueryAsync<dynamic>(cnn, typeof(DapperRow), new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered, default(CancellationToken)));
+        }
+
+        /// <summary>
+        /// Execute a query asynchronously using .NET 4.5 Task.
+        /// </summary>
+        public static Task<IEnumerable<dynamic>> QueryAsync(this IDbConnection cnn, CommandDefinition command)
+        {
+            return QueryAsync<dynamic>(cnn, typeof(DapperRow), command);
+        }
+
         /// <summary>
         /// Execute a query asynchronously using .NET 4.5 Task.
         /// </summary>
@@ -52,19 +69,49 @@ private static async Task<IEnumerable<T>> QueryAsync<T>(this IDbConnection cnn,
             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;
+            var cancel = command.CancellationToken;
             using (var cmd = (DbCommand)command.SetupCommand(cnn, info.ParamReader))
             {
+                DbDataReader reader = null;
                 try
                 {
-                    if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false);
-                    using (var reader = await cmd.ExecuteReaderAsync(command.CancellationToken).ConfigureAwait(false))
+                    if (wasClosed) await ((DbConnection)cnn).OpenAsync(cancel).ConfigureAwait(false);
+                    reader = await cmd.ExecuteReaderAsync(wasClosed ? CommandBehavior.CloseConnection | CommandBehavior.SequentialAccess : CommandBehavior.SequentialAccess, cancel).ConfigureAwait(false);
+                    
+                    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 (command.AddToCache) SetQueryCache(identity, info);
+                    }
+
+                    var func = tuple.Func;
+
+                    if (command.Buffered)
+                    {
+                        List<T> buffer = new List<T>();
+                        while (await reader.ReadAsync(cancel).ConfigureAwait(false))
+                        {
+                            buffer.Add((T)func(reader));
+                        }
+                        return buffer;
+                    }
+                    else
                     {
-                        return ExecuteReader<T>(reader, effectiveType, identity, info, command.AddToCache).ToList();
+                        // can't use ReadAsync / cancellation; but this will have to do
+                        wasClosed = false; // don't close if handing back an open reader; rely on the command-behavior
+                        var deferred = ExecuteReaderSync<T>(reader, func);
+                        reader = null; // to prevent it being disposed before the caller gets to see it
+                        return deferred;
                     }
+                    
                 }
                 finally
                 {
+                    using (reader) { } // dispose if non-null
                     if (wasClosed) cnn.Close();
+                    
                 }
             }
         }
@@ -111,7 +158,7 @@ private static async Task<int> ExecuteMultiImplAsync(IDbConnection cnn, CommandD
             bool wasClosed = cnn.State == ConnectionState.Closed;
             try
             {
-                if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false);
+                if (wasClosed) await ((DbConnection)cnn).OpenAsync(command.CancellationToken).ConfigureAwait(false);
                 
                 CacheInfo info = null;
                 string masterSql = null;
@@ -205,7 +252,7 @@ private static async Task<int> ExecuteImplAsync(IDbConnection cnn, CommandDefini
             {
                 try
                 {
-                    if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false);
+                    if (wasClosed) await ((DbConnection)cnn).OpenAsync(command.CancellationToken).ConfigureAwait(false);
                     return await cmd.ExecuteNonQueryAsync(command.CancellationToken).ConfigureAwait(false);
                 }
                 finally
@@ -394,10 +441,11 @@ private static async Task<int> ExecuteImplAsync(IDbConnection cnn, CommandDefini
             bool wasClosed = cnn.State == ConnectionState.Closed;
             try
             {
-                if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false);
+                if (wasClosed) await ((DbConnection)cnn).OpenAsync(command.CancellationToken).ConfigureAwait(false);
                 using (var cmd = (DbCommand)command.SetupCommand(cnn, info.ParamReader))
-                using (var reader = await cmd.ExecuteReaderAsync(command.CancellationToken).ConfigureAwait(false))
+                using (var reader = await cmd.ExecuteReaderAsync(wasClosed ? CommandBehavior.CloseConnection | CommandBehavior.SequentialAccess : CommandBehavior.SequentialAccess, command.CancellationToken).ConfigureAwait(false))
                 {
+                    if (!command.Buffered) wasClosed = false; // handing back open reader; rely on command-behavior
                     var results = MultiMapImpl<TFirst, TSecond, TThird, TFourth, TFifth, TSixth, TSeventh, TReturn>(null, default(CommandDefinition), map, splitOn, reader, identity);
                     return command.Buffered ? results.ToList() : results;
                 }
@@ -407,18 +455,8 @@ private static async Task<int> ExecuteImplAsync(IDbConnection cnn, CommandDefini
             }
         }
 
-        private static IEnumerable<T> ExecuteReader<T>(IDataReader reader, Type effectiveType, Identity identity, CacheInfo info, bool addToCache)
+        private static IEnumerable<T> ExecuteReaderSync<T>(IDataReader reader, Func<IDataReader,object> func)
         {
-            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);
@@ -439,6 +477,107 @@ private static IEnumerable<T> ExecuteReader<T>(IDataReader reader, Type effectiv
             var command = new CommandDefinition(sql, (object)param, transaction, commandTimeout, commandType, CommandFlags.Buffered);
             return QueryMultipleAsync(cnn, command);
         }
+
+        partial class GridReader
+        {
+            CancellationToken cancel;
+            internal GridReader(IDbCommand command, IDataReader reader, Identity identity, CancellationToken cancel) : this(command, reader, identity)
+            {
+                this.cancel = cancel;
+            }
+
+            /// <summary>
+            /// Read the next grid of results, returned as a dynamic object
+            /// </summary>
+            public Task<IEnumerable<dynamic>> ReadAsync(bool buffered = true)
+            {
+                return ReadAsyncImpl<dynamic>(typeof(DapperRow), buffered);
+            }
+
+            /// <summary>
+            /// Read the next grid of results
+            /// </summary>
+            public Task<IEnumerable<object>> ReadAsync(Type type, bool buffered = true)
+            {
+                if (type == null) throw new ArgumentNullException("type");
+                return ReadAsyncImpl<object>(type, buffered);
+            }
+            /// <summary>
+            /// Read the next grid of results
+            /// </summary>
+            public Task<IEnumerable<T>> ReadAsync<T>(bool buffered = true)
+            {
+                return ReadAsyncImpl<T>(typeof(T), buffered);
+            }
+
+            private async Task NextResultAsync()
+            {
+                if (await ((DbDataReader)reader).NextResultAsync(cancel).ConfigureAwait(false))
+                {
+                    readCount++;
+                    gridIndex++;
+                    consumed = false;
+                }
+                else
+                {
+                    // happy path; close the reader cleanly - no
+                    // need for "Cancel" etc
+                    reader.Dispose();
+                    reader = null;
+
+                    Dispose();
+                }
+            }
+
+            private Task<IEnumerable<T>> ReadAsyncImpl<T>(Type type, bool buffered)
+            {
+                if (reader == null) throw new ObjectDisposedException(GetType().FullName, "The reader has been disposed; this can happen after all data has been consumed");
+                if (consumed) throw new InvalidOperationException("Query results must be consumed in the correct order, and each result can only be consumed once");
+                var typedIdentity = identity.ForGrid(type, gridIndex);
+                CacheInfo cache = GetCacheInfo(typedIdentity, null, true);
+                var deserializer = cache.Deserializer;
+
+                int hash = GetColumnHash(reader);
+                if (deserializer.Func == null || deserializer.Hash != hash)
+                {
+                    deserializer = new DeserializerState(hash, GetDeserializer(type, reader, 0, -1, false));
+                    cache.Deserializer = deserializer;
+                }
+                consumed = true;
+                if (buffered && this.reader is DbDataReader)
+                {
+                    return ReadBufferedAsync<T>(gridIndex, deserializer.Func, typedIdentity);
+                }
+                else
+                {
+                    var result = ReadDeferred<T>(gridIndex, deserializer.Func, typedIdentity);
+                    if (buffered) result = result.ToList(); // for the "not a DbDataReader" scenario
+                    return Task.FromResult(result);
+                }
+            }
+
+            private async Task<IEnumerable<T>> ReadBufferedAsync<T>(int index, Func<IDataReader, object> deserializer, Identity typedIdentity)
+            {
+                try
+                {
+                    var reader = (DbDataReader)this.reader;
+                    List<T> buffer = new List<T>();
+                    while (index == gridIndex && await reader.ReadAsync(cancel).ConfigureAwait(false))
+                    {
+                        buffer.Add((T)deserializer(reader));
+                    }
+                    return buffer;
+                }
+                finally // finally so that First etc progresses things even when multiple rows
+                {
+                    if (index == gridIndex)
+                    {
+                        await NextResultAsync().ConfigureAwait(false);
+                    }
+                }
+            }
+        }
+
         /// <summary>
         /// Execute a command that returns multiple result sets, and access each in turn
         /// </summary>
@@ -453,11 +592,11 @@ public static async Task<GridReader> QueryMultipleAsync(this IDbConnection cnn,
             bool wasClosed = cnn.State == ConnectionState.Closed;
             try
             {
-                if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false);
+                if (wasClosed) await ((DbConnection)cnn).OpenAsync(command.CancellationToken).ConfigureAwait(false);
                 cmd = (DbCommand)command.SetupCommand(cnn, info.ParamReader);
-                reader = await cmd.ExecuteReaderAsync(wasClosed ? CommandBehavior.CloseConnection : CommandBehavior.Default, command.CancellationToken).ConfigureAwait(false);
+                reader = await cmd.ExecuteReaderAsync(wasClosed ? CommandBehavior.CloseConnection | CommandBehavior.SequentialAccess : CommandBehavior.SequentialAccess, command.CancellationToken).ConfigureAwait(false);
 
-                var result = new GridReader(cmd, reader, identity);
+                var result = new GridReader(cmd, reader, identity, command.CancellationToken);
                 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
@@ -535,8 +674,8 @@ private static async Task<IDataReader> ExecuteReaderImplAsync(IDbConnection cnn,
             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);
+                if (wasClosed) await ((DbConnection)cnn).OpenAsync(command.CancellationToken).ConfigureAwait(false);
+                var reader = await cmd.ExecuteReaderAsync(wasClosed ? CommandBehavior.CloseConnection | CommandBehavior.SequentialAccess : CommandBehavior.SequentialAccess, command.CancellationToken).ConfigureAwait(false);
                 wasClosed = false;
                 return reader;
             }
@@ -613,7 +752,7 @@ private async static Task<T> ExecuteScalarImplAsync<T>(IDbConnection cnn, Comman
             try
             {
                 cmd = (DbCommand)command.SetupCommand(cnn, paramReader);
-                if (wasClosed) await ((DbConnection)cnn).OpenAsync().ConfigureAwait(false);
+                if (wasClosed) await ((DbConnection)cnn).OpenAsync(command.CancellationToken).ConfigureAwait(false);
                 result = await cmd.ExecuteScalarAsync(command.CancellationToken).ConfigureAwait(false);
             }
             finally

Dapper.Contrib.Tests/Properties/AssemblyInfo.cs

@@ -31,5 +31,5 @@
 //
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("1.29.0.0")]
-[assembly: AssemblyFileVersion("1.29.0.0")]
+[assembly: AssemblyVersion("1.30.0.0")]
+[assembly: AssemblyFileVersion("1.30.0.0")]

Dapper.Contrib/Properties/AssemblyInfo.cs

@@ -31,5 +31,5 @@
 //
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("1.29.0.0")]
-[assembly: AssemblyFileVersion("1.29.0.0")]
+[assembly: AssemblyVersion("1.30.0.0")]
+[assembly: AssemblyFileVersion("1.30.0.0")]

Dapper.SqlBuilder/Properties/AssemblyInfo.cs

@@ -31,5 +31,5 @@
 //
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("1.29.0.0")]
-[assembly: AssemblyFileVersion("1.29.0.0")]
+[assembly: AssemblyVersion("1.30.0.0")]
+[assembly: AssemblyFileVersion("1.30.0.0")]

DapperTests NET35/Properties/AssemblyInfo.cs

@@ -31,5 +31,5 @@
 //
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("1.29.0.0")]
-[assembly: AssemblyFileVersion("1.29.0.0")]
+[assembly: AssemblyVersion("1.30.0.0")]
+[assembly: AssemblyFileVersion("1.30.0.0")]

DapperTests NET45/Properties/AssemblyInfo.cs

@@ -31,5 +31,5 @@
 //
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("1.29.0.0")]
-[assembly: AssemblyFileVersion("1.29.0.0")]
+[assembly: AssemblyVersion("1.30.0.0")]
+[assembly: AssemblyFileVersion("1.30.0.0")]

DapperTests NET45/Tests.cs

@@ -4,6 +4,9 @@
 using System.Data;
 using System.Diagnostics;
 using System;
+using System.Threading.Tasks;
+using System.Threading;
+using System.Data.SqlClient;
 
 namespace DapperTests_NET45
 {
@@ -18,6 +21,35 @@ public void TestBasicStringUsageAsync()
                 arr.IsSequenceEqualTo(new[] { "abc", "def" });
             }
         }
+        public void TestBasicStringUsageAsyncNonBuffered()
+        {
+            using (var connection = Program.GetOpenConnection())
+            {
+                var query = connection.QueryAsync<string>(new CommandDefinition("select 'abc' as [Value] union all select @txt", new { txt = "def" }, flags:  CommandFlags.None));
+                var arr = query.Result.ToArray();
+                arr.IsSequenceEqualTo(new[] { "abc", "def" });
+            }
+        }
+
+        public void TestLongOperationWithCancellation()
+        {
+            using(var connection = Program.GetClosedConnection())
+            {
+                CancellationTokenSource cancel = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+                var task = connection.QueryAsync<int>(new CommandDefinition("waitfor delay '00:00:10';select 1", cancellationToken: cancel.Token));
+                try
+                {
+                    if (!task.Wait(TimeSpan.FromSeconds(7)))
+                    {
+                        throw new TimeoutException(); // should have cancelled
+                    }
+                }
+                catch (AggregateException agg)
+                {
+                    (agg.InnerException is SqlException).IsTrue();
+                }
+            }
+        }
 
         public void TestBasicStringUsageClosedAsync()
         {
@@ -29,6 +61,16 @@ public void TestBasicStringUsageClosedAsync()
             }
         }
 
+        public void TestQueryDynamicAsync()
+        {
+            using (var connection = Program.GetClosedConnection())
+            {
+                var row = connection.QueryAsync("select 'abc' as [Value]").Result.Single();
+                string value = row.Value;
+                value.IsEqualTo("abc");
+            }
+        }
+
         public void TestClassWithStringUsageAsync()
         {
             using (var connection = Program.GetOpenConnection())
@@ -104,8 +146,8 @@ public void TestMultiAsync()
             {
                 using (Dapper.SqlMapper.GridReader multi = conn.QueryMultipleAsync("select 1; select 2").Result)
                 {
-                    multi.Read<int>().Single().IsEqualTo(1);
-                    multi.Read<int>().Single().IsEqualTo(2);
+                    multi.ReadAsync<int>().Result.Single().IsEqualTo(1);
+                    multi.ReadAsync<int>().Result.Single().IsEqualTo(2);
                 }
             }
         }
@@ -115,8 +157,8 @@ public void TestMultiClosedConnAsync()
             {
                 using (Dapper.SqlMapper.GridReader multi = conn.QueryMultipleAsync("select 1; select 2").Result)
                 {
-                    multi.Read<int>().Single().IsEqualTo(1);
-                    multi.Read<int>().Single().IsEqualTo(2);
+                    multi.ReadAsync<int>().Result.Single().IsEqualTo(1);
+                    multi.ReadAsync<int>().Result.Single().IsEqualTo(2);
                 }
             }
         }

Tests/Properties/AssemblyInfo.cs

@@ -31,5 +31,5 @@
 //
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
-[assembly: AssemblyVersion("1.29.0.0")]
-[assembly: AssemblyFileVersion("1.29.0.0")]
+[assembly: AssemblyVersion("1.30.0.0")]
+[assembly: AssemblyFileVersion("1.30.0.0")]

dapper.nuspec

@@ -2,7 +2,7 @@
 <package xmlns="http://schemas.microsoft.com/packaging/2010/07/nuspec.xsd">
   <metadata schemaVersion="2">
     <id>Dapper</id>
-    <version>1.29</version>
+    <version>1.30</version>
     <title>Dapper dot net</title>
     <authors>Sam Saffron,Marc Gravell</authors>
     <owners>Sam Saffron,Marc Gravell</owners>
@@ -19,6 +19,7 @@
       <frameworkAssembly assemblyName="Microsoft.CSharp" targetFramework=".NETFramework4.0-Client, .NETFramework4.0" />
     </frameworkAssemblies>
     <releaseNotes>
+      * 1.30    - Better async cancellation
       * 1.29    - Make underscore name matching optional (opt-in) - this can be a breaking change for some people
       * 1.28    - Much better numeric type conversion; fix for large oracle strings; map Foo_Bar to FooBar (etc); ExecuteScalar added; stability fixes
       * 1.27    - Fixes for type-handler parse; ensure type-handlers get last dibs on configuring parameters