Add support for using cancellation tokens to cancel asynchronous queries. Additional tests.

Author: bordoley

SQLitePCL.pretty.Async/AsyncDatabaseConnection.cs

@@ -33,6 +33,11 @@ namespace SQLitePCL.pretty
     /// </summary>
     public static partial class DatabaseConnection
     {
+        // FIXME: I picked this number fairly randomly. It would be good to do some experimentation
+        // to determine if its a good default. The goal should be supporting cancelling of queries that are
+        // actually blocking use of the database for a measurable period of time.
+        private static readonly int defaultInterruptInstructionCount = 100;
+
         private static IScheduler defaultScheduler = TaskPoolScheduler.Default;
 
         /// <summary>
@@ -54,6 +59,13 @@ public static IScheduler DefaultScheduler
             }
         }
 
+        internal static IAsyncDatabaseConnection AsAsyncDatabaseConnection(this SQLiteDatabaseConnection This, IScheduler scheduler, int interruptInstructionCount)
+        {
+            Contract.Requires(This != null);
+            Contract.Requires(scheduler != null);
+            return new AsyncDatabaseConnectionImpl(This, scheduler, interruptInstructionCount);
+        }
+
         /// <summary>
         /// Returns an <see cref="IAsyncDatabaseConnection"/> instance that delegates database requests
         /// to the provided <see cref="IDatabaseConnection"/>.
@@ -66,9 +78,7 @@ public static IScheduler DefaultScheduler
         /// <returns>An <see cref="IAsyncDatabaseConnection"/> instance.</returns>
         public static IAsyncDatabaseConnection AsAsyncDatabaseConnection(this SQLiteDatabaseConnection This, IScheduler scheduler)
         {
-            Contract.Requires(This != null);
-            Contract.Requires(scheduler != null);
-            return new AsyncDatabaseConnectionImpl(This, scheduler);
+            return AsAsyncDatabaseConnection(This, scheduler, defaultInterruptInstructionCount);
         }
 
         /// <summary>
@@ -82,7 +92,6 @@ public static IAsyncDatabaseConnection AsAsyncDatabaseConnection(this SQLiteData
         /// <returns>An <see cref="IAsyncDatabaseConnection"/> instance.</returns>
         public static IAsyncDatabaseConnection AsAsyncDatabaseConnection(this SQLiteDatabaseConnection This)
         {
-            Contract.Requires(This != null);
             return AsAsyncDatabaseConnection(This, defaultScheduler);
         }
     }
@@ -520,6 +529,7 @@ internal sealed class AsyncDatabaseConnectionImpl : IAsyncDatabaseConnection
     {
         private readonly OperationsQueue queue = new OperationsQueue();
         private readonly IScheduler scheduler;
+        private readonly int interruptInstructionCount;
 
         private readonly SQLiteDatabaseConnection conn;
         private readonly IObservable<DatabaseTraceEventArgs> trace;
@@ -528,10 +538,11 @@ internal sealed class AsyncDatabaseConnectionImpl : IAsyncDatabaseConnection
 
         private bool disposed = false;
 
-        internal AsyncDatabaseConnectionImpl(SQLiteDatabaseConnection conn, IScheduler scheduler)
+        internal AsyncDatabaseConnectionImpl(SQLiteDatabaseConnection conn, IScheduler scheduler, int interruptInstructionCount)
         {
             this.conn = conn;
             this.scheduler = scheduler;
+            this.interruptInstructionCount = interruptInstructionCount;
 
             this.trace = Observable.FromEventPattern<DatabaseTraceEventArgs>(conn, "Trace").Select(e => e.EventArgs);
             this.profile = Observable.FromEventPattern<DatabaseProfileEventArgs>(conn, "Profile").Select(e => e.EventArgs);
@@ -632,7 +643,7 @@ public IObservable<T> Use<T>(Func<IDatabaseConnection, CancellationToken, IEnume
 
                     return queue.EnqueueOperation(ct =>
                         {
-                            //this.conn.RegisterProgressHandler(100, () => ct.IsCancellationRequested);
+                            this.conn.RegisterProgressHandler(interruptInstructionCount, () => ct.IsCancellationRequested);
                             try
                             {
                                 ct.ThrowIfCancellationRequested();
@@ -653,7 +664,7 @@ public IObservable<T> Use<T>(Func<IDatabaseConnection, CancellationToken, IEnume
                             }
                             finally
                             {
-                                //this.conn.RemoveProgressHandler();
+                                this.conn.RemoveProgressHandler();
                             }
                         }, scheduler, cancellationToken);
                 });

SQLitePCL.pretty.Async/OperationsQueue.cs

@@ -121,8 +121,8 @@ internal static class OperationsQueueExtensions
         public static Task EnqueueOperation(this OperationsQueue This, Action<CancellationToken> calculationFunc, IScheduler scheduler, CancellationToken cancellationToken)
         {
             return This.EnqueueOperation(ct =>
-                Observable.Start(() => 
-                    calculationFunc(ct), scheduler).ToTask(ct), cancellationToken);
+                Observable.Start(() => calculationFunc(ct), scheduler).ToTask(), 
+                cancellationToken);
         }
     }
 }

SQLitePCL.pretty.tests/AsyncTests/AsyncDatabaseConnectionTests.cs

@@ -19,6 +19,7 @@ limitations under the License.
 using System.Collections.Generic;
 using System.Linq;
 using System.Reactive;
+using System.Reactive.Concurrency;
 using System.Reactive.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -251,6 +252,11 @@ await adb.Use(db =>
                     Assert.DoesNotThrow(() => { var x = db.IsDatabaseReadOnly("main"); });
                     Assert.DoesNotThrow(() => { using (var stmt = db.PrepareStatement("SELECT x FROM foo;")) { } });
 
+                    int current;
+                    int highwater;
+                    Assert.DoesNotThrow(() => { db.Status(DatabaseConnectionStatusCode.CacheMiss, out current, out highwater, false); });
+                    Assert.DoesNotThrow(() => { db.WalCheckPoint("main"); });
+
                     db.Dispose();
 
                     Assert.Throws<ObjectDisposedException>(() => { var x = db.IsAutoCommit; });
@@ -261,7 +267,10 @@ await adb.Use(db =>
                     Assert.Throws<ObjectDisposedException>(() => { var x = db.Statements; });
                     Assert.Throws<ObjectDisposedException>(() => { var x = db.IsDatabaseReadOnly("main"); });
                     Assert.Throws<ObjectDisposedException>(() => { var x = db.OpenBlob("", "", "", 0, true); });
+                    Assert.Throws<ObjectDisposedException>(() => { var x = db.GetTableColumnMetadata("", "", ""); });
                     Assert.Throws<ObjectDisposedException>(() => db.PrepareStatement("SELECT x FROM foo;"));
+                    Assert.Throws<ObjectDisposedException>(() => { db.Status(DatabaseConnectionStatusCode.CacheMiss, out current, out highwater, false); });
+                    Assert.Throws<ObjectDisposedException>(() => { db.WalCheckPoint("main"); });
                 });
 
                 await adb.Use(db =>
@@ -338,5 +347,21 @@ await adb.Query("Select ?, ?, ?", _0, _1, _2)
                 Assert.AreEqual(result.Item3 != 0, _2);
             }
         }
+
+        [Test]
+        public void TestStatementCancellation()
+        {
+            using (var adb = SQLite3.Open(":memory:").AsAsyncDatabaseConnection(TaskPoolScheduler.Default, 1))
+            {
+                var cts = new CancellationTokenSource();
+                Assert.That(async () =>
+                    await adb.Use((db, ct) => 
+                        {
+                            cts.Cancel();
+                            db.Execute("Select 1;");
+                            Assert.Fail();
+                        }, cts.Token), Throws.TypeOf<TaskCanceledException>());
+            }
+        }
     }
 }

SQLitePCL.pretty.tests/AsyncTests/AsyncStatementTests.cs

@@ -43,6 +43,7 @@ await aStmt.Use(stmt =>
                         Assert.Throws<ObjectDisposedException>(() => { stmt.ClearBindings(); });
                         Assert.Throws<ObjectDisposedException>(() => { stmt.MoveNext(); });
                         Assert.Throws<ObjectDisposedException>(() => { stmt.Reset(); });
+                        Assert.Throws<ObjectDisposedException>(() => { stmt.Status(StatementStatusCode.AutoIndex, false); });
                     });
 
                 await aStmt.Use(stmt =>