Refactor database configuration into a builder class and eliminate the need for expando's and other weird static state issues

Author: bordoley

SQLitePCL.pretty.Async/AsyncDatabaseConnection.cs

@@ -25,89 +25,71 @@ limitations under the License.
 using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Reactive.Subjects;
 
 namespace SQLitePCL.pretty
 {
-    /// <summary>
-    /// Extensions methods for <see cref="IDatabaseConnection"/>
-    /// </summary>
-    public static partial class DatabaseConnection
+    internal class WriteLockedRef<T>
     {
-        private const string AsyncDatabaseConnectionKey = "AsyncDatabaseConnection";
+        private readonly object gate = new object();
+        private T value;
 
-        // 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 volatile IScheduler defaultScheduler = TaskPoolScheduler.Default;
+        public WriteLockedRef(T defaultValue)
+        {
+            this.value = defaultValue;
+        }
 
-        /// <summary>
-        /// Allows an application to set a default scheduler for <see cref="IAsyncDatabaseConnection"/>
-        /// instances created with <see cref="DatabaseConnection.AsAsyncDatabaseConnection(SQLiteDatabaseConnection)"/>.
-        /// </summary>
-        /// <remarks>This is a convenience feature that allows an application to set a global
-        /// <see cref="IScheduler"/> instance, instead of supplying it with each call to
-        /// <see cref="DatabaseConnection.AsAsyncDatabaseConnection(SQLiteDatabaseConnection)"/>.
-        /// </remarks>
-        /// <threadsafety static="false">This setter sets global state and should not be
-        /// used after application initialization.</threadsafety>
-        public static IScheduler DefaultScheduler
+        public T Value
         {
-            set
+            get { return value; }
+
+            set 
             {
-                Contract.Requires(value != null);
-                defaultScheduler = value;
+                lock (gate)
+                {
+                    this.value = value;
+                }
             }
         }
-        
-        internal static IAsyncDatabaseConnection AsAsyncDatabaseConnection(this SQLiteDatabaseConnection This, IScheduler scheduler, int interruptInstructionCount)
+
+    }
+
+    /// <summary>
+    /// SQLiteDatabaseConnectionBuilder extension functions.
+    /// </summary>
+    public static class SQLiteDatabaseConnectionBuilderExtensions
+    {
+        /// <summary>
+        /// Builds an IAsyncDatabaseConnection using the specified scheduler.
+        /// </summary>
+        /// <returns>An IAsyncDatabaseConnection using the specified scheduler.</returns>
+        /// <param name="This">A SQLiteDatabaseConnectionBuilder instance.</param>
+        /// <param name="scheduler">An RX scheduler</param>
+        public static IAsyncDatabaseConnection BuildAsyncDatabaseConnection(
+            this SQLiteDatabaseConnectionBuilder This,
+            IScheduler scheduler)
         {
             Contract.Requires(This != null);
             Contract.Requires(scheduler != null);
 
-            IAsyncDatabaseConnection target;
-            if (DatabaseConnectionExpando.Instance.GetOrAddValue(This, AsyncDatabaseConnectionKey, _ =>
-                {
-                    // Store a WeakReference to the async connection so that we don't end up with a circular reference that prevents the 
-                    // SQLiteDatabaseConnection from being freed. 
-                    var asyncConnection = new AsyncDatabaseConnectionImpl(This, scheduler, interruptInstructionCount);
-                    return new WeakReference<IAsyncDatabaseConnection>(asyncConnection);
-                }).TryGetTarget(out target))
-            {
-                return target;
-            }
+            var builder = This.Clone();
 
-            // This can't really happen. 
-            throw new InvalidOperationException();
-        }
+            var progressHandlerResult = new WriteLockedRef<bool>(false);
+            builder.ProgressHandler = () => progressHandlerResult.Value;
+            var db = This.Build();
 
-        /// <summary>
-        /// Returns an <see cref="IAsyncDatabaseConnection"/> instance that delegates database requests
-        /// to the provided <see cref="IDatabaseConnection"/>.
-        /// </summary>
-        /// <remarks>Note, once this method is called the provided <see cref="IDatabaseConnection"/>
-        /// is owned by the returned <see cref="IAsyncDatabaseConnection"/>, and may no longer be
-        /// safely used directly.</remarks>
-        /// <param name="This">The database connection.</param>
-        /// <param name="scheduler">A scheduler used to schedule asynchronous database use on.</param>
-        /// <returns>An <see cref="IAsyncDatabaseConnection"/> instance.</returns>
-        public static IAsyncDatabaseConnection AsAsyncDatabaseConnection(this SQLiteDatabaseConnection This, IScheduler scheduler) =>
-            AsAsyncDatabaseConnection(This, scheduler, defaultInterruptInstructionCount);
+            return new AsyncDatabaseConnectionImpl(db, scheduler, progressHandlerResult);
+        }
 
         /// <summary>
-        /// Returns an <see cref="IAsyncDatabaseConnection"/> instance that delegates database requests
-        /// to the provided <see cref="IDatabaseConnection"/>.
+        /// Builds an IAsyncDatabaseConnection using the default TaskPool scheduler.
         /// </summary>
-        /// <remarks>Note, once this method is called the provided <see cref="IDatabaseConnection"/>
-        /// is owned by the returned <see cref="IAsyncDatabaseConnection"/>, and may no longer be
-        /// safely used directly.</remarks>
-        /// <param name="This">The database connection.</param>
-        /// <returns>An <see cref="IAsyncDatabaseConnection"/> instance.</returns>
-        public static IAsyncDatabaseConnection AsAsyncDatabaseConnection(this SQLiteDatabaseConnection This) =>
-            AsAsyncDatabaseConnection(This, defaultScheduler);
+        /// <returns>An IAsyncDatabaseConnection using the default TaskPool scheduler.</returns>
+        /// <param name="This">A SQLiteDatabaseConnectionBuilder instance.</param>
+        public static IAsyncDatabaseConnection BuildAsyncDatabaseConnection(this SQLiteDatabaseConnectionBuilder This) =>
+            This.BuildAsyncDatabaseConnection(TaskPoolScheduler.Default);
     }
-
+        
     /// <summary>
     /// Extensions methods for <see cref="IAsyncDatabaseConnection"/>.
     /// </summary>
@@ -491,19 +473,28 @@ 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 WriteLockedRef<bool> progressHandlerResult;
 
         private bool disposed = false;
 
-        internal AsyncDatabaseConnectionImpl(SQLiteDatabaseConnection conn, IScheduler scheduler, int interruptInstructionCount)
+        internal AsyncDatabaseConnectionImpl(SQLiteDatabaseConnection conn, IScheduler scheduler, WriteLockedRef<bool> progressHandlerResult)
         {
             this.conn = conn;
             this.scheduler = scheduler;
-            this.interruptInstructionCount = interruptInstructionCount;
+            this.progressHandlerResult = progressHandlerResult;
+
+            this.Trace = Observable.FromEventPattern<DatabaseTraceEventArgs>(conn, "Trace").Select(e => e.EventArgs);
+            this.Profile = Observable.FromEventPattern<DatabaseProfileEventArgs>(conn, "Profile").Select(e => e.EventArgs);
+            this.Update = Observable.FromEventPattern<DatabaseUpdateEventArgs>(conn, "Update").Select(e => e.EventArgs);
         }
 
+        public IObservable<DatabaseTraceEventArgs> Trace { get; }
+
+        public IObservable<DatabaseProfileEventArgs> Profile { get; }
+
+        public IObservable<DatabaseUpdateEventArgs> Update { get; }
+
         public async Task DisposeAsync()
         {
             if (disposed)
@@ -576,7 +567,9 @@ public IObservable<T> Use<T>(Func<IDatabaseConnection, CancellationToken, IEnume
 
                     return queue.EnqueueOperation(ct =>
                         {
-                            this.conn.RegisterProgressHandler(interruptInstructionCount, () => ct.IsCancellationRequested);
+                            this.progressHandlerResult.Value = false;
+                            var ctSubscription = ct.Register(() => this.progressHandlerResult.Value = true);
+                                
                             try
                             {
                                 ct.ThrowIfCancellationRequested();
@@ -597,7 +590,8 @@ public IObservable<T> Use<T>(Func<IDatabaseConnection, CancellationToken, IEnume
                             }
                             finally
                             {
-                                this.conn.RemoveProgressHandler();
+                                ctSubscription.Dispose();
+                                this.progressHandlerResult.Value = false;
                             }
                         }, scheduler, cancellationToken);
                 });

SQLitePCL.pretty.Async/Interfaces.cs

@@ -26,6 +26,27 @@ namespace SQLitePCL.pretty
     /// </summary>
     public interface IAsyncDatabaseConnection : IDisposable
     {
+        /// <summary>
+        /// A hot <see cref="IObservable&lt;DatabaseTraceEventArgs&gt;"/> of this connection's SQLite trace events.
+        /// </summary>
+        /// <seealso cref="SQLiteDatabaseConnection.Trace"/>
+        /// <seealso href="https://sqlite.org/c3ref/profile.html"/>
+        IObservable<DatabaseTraceEventArgs> Trace { get; }
+
+        /// <summary>
+        /// A hot <see cref="IObservable&lt;DatabaseProfileEventArgs&gt;"/> of this connection's SQLite profile events.
+        /// </summary>
+        /// /// <seealso cref="SQLiteDatabaseConnection.Profile"/>
+        /// <seealso href="https://sqlite.org/c3ref/profile.html"/>
+        IObservable<DatabaseProfileEventArgs> Profile { get; }
+
+        /// <summary>
+        /// A hot <see cref="IObservable&lt;DatabaseUpdateEventArgs&gt;"/> of this connection's SQLite update events.
+        /// </summary>
+        /// /// <seealso cref="SQLiteDatabaseConnection.Update"/>
+        /// <seealso href="https://sqlite.org/c3ref/update_hook.html"/>
+        IObservable<DatabaseUpdateEventArgs> Update { get; }
+
         /// <summary>
         /// Shutdown the underlying operations queue used by the <see cref="IAsyncDatabaseConnection"/>
         /// and prevents the queuing of additional database access requests. Requests to access the database

SQLitePCL.pretty.tests/AsyncTests/AsyncBlobStreamTests.cs

@@ -28,7 +28,7 @@ public class AsyncBlobStreamTests
         [Fact]
         public async Task TestDispose()
         {
-            using (var db = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var db = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 await db.ExecuteAsync("CREATE TABLE foo (x blob);");
                 await db.ExecuteAsync("INSERT INTO foo (x) VALUES(?);", "data");
@@ -53,7 +53,7 @@ await db.Query("SELECT rowid, x FROM foo")
         [Fact]
         public async Task TestRead()
         {
-            using (var db = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var db = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 byte[] bytes = new byte[1000];
                 Random random = new Random();
@@ -89,7 +89,7 @@ public async Task TestRead()
         [Fact]
         public async Task TestWrite()
         {
-            using (var db = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var db = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 byte[] bytes = new byte[1000];
                 Random random = new Random();
@@ -138,7 +138,7 @@ await db.Query("SELECT rowid, x FROM foo")
         [Fact]
         public async Task TestSeek()
         {
-            using (var db = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var db = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 await db.ExecuteAsync("CREATE TABLE foo (x blob);");
                 await db.ExecuteAsync("INSERT INTO foo (x) VALUES(?);", "data");

SQLitePCL.pretty.tests/AsyncTests/AsyncDatabaseConnectionTests.cs

@@ -28,10 +28,78 @@ namespace SQLitePCL.pretty.tests
 {
     public class AsyncDatabaseConnectionTests
     {
+        [Fact]
+        public async Task TestProfileEvent()
+        {
+            using (var db = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
+            {
+                var statement = "CREATE TABLE foo (x int);";
+                db.Profile.Subscribe(e =>
+                    {
+                        Assert.Equal(statement, e.Statement);
+                        Assert.True(TimeSpan.MinValue < e.ExecutionTime);
+                    });
+
+                await db.ExecuteAsync(statement);
+            }
+        }
+
+        [Fact]
+        public async Task TestTraceEvent()
+        {
+            using (var db = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
+            {
+                var statement = "CREATE TABLE foo (x int);";
+                db.Trace.Subscribe(e =>
+                    {
+                        Assert.Equal(statement, e.Statement);
+                    });
+
+                await db.ExecuteAsync(statement);
+
+                statement = "INSERT INTO foo (x) VALUES (1);";
+                await db.ExecuteAsync(statement);
+            }
+        }
+
+        [Fact]
+        public async Task TestUpdateEvent()
+        {
+            using (var db = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
+            {
+                var currentAction = ActionCode.CreateTable;
+                var rowid = 1;
+
+                db.Update.Subscribe(e =>
+                    {
+                        Assert.Equal(currentAction, e.Action);
+                        Assert.Equal("main", e.Database);
+                        Assert.Equal("foo", e.Table);
+                        Assert.Equal(rowid, e.RowId);
+                    });
+
+                currentAction = ActionCode.CreateTable;
+                rowid = 1;
+                await db.ExecuteAsync("CREATE TABLE foo (x int);");
+
+                currentAction = ActionCode.Insert;
+                rowid = 1;
+                await db.ExecuteAsync("INSERT INTO foo (x) VALUES (1);");
+
+                currentAction = ActionCode.Insert;
+                rowid = 2;
+                await db.ExecuteAsync("INSERT INTO foo (x) VALUES (2);");
+
+                currentAction = ActionCode.DropTable;
+                rowid = 2;
+                await db.ExecuteAsync("DROP TABLE foo");
+            }
+        }
+
         [Fact]
         public async Task TestUse()
         {
-            using (var adb = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var adb = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 await adb.Use(db => Enumerable.Range(0, 1000))
                     .Scan(Tuple.Create(-1, -1), (x, y) => Tuple.Create(x.Item1 + 1, y))
@@ -63,7 +131,7 @@ await adb.Use(db => Enumerable.Range(0, 1000))
         [Fact]
         public async Task TestIDatabaseConnectionDispose()
         {
-            using (var adb = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var adb = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 await adb.ExecuteAsync("CREATE TABLE foo (x int);");
 
@@ -124,7 +192,7 @@ await disposedObservable.Materialize()
         [Fact]
         public void TestUseCancelled()
         {
-            using (var adb = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var adb = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 var cts = new CancellationTokenSource();
                 cts.Cancel();
@@ -138,7 +206,7 @@ public void TestUseCancelled()
         [Fact]
         public async Task TestPrepareAllAsync()
         {
-            using (var adb = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var adb = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 await adb.ExecuteAsync("CREATE TABLE foo (x int);");
                 var stmts =
@@ -161,7 +229,7 @@ await adb.PrepareAllAsync(
         [Fact]
         public async Task TestQuery()
         {
-            using (var adb = SQLite3.OpenInMemory().AsAsyncDatabaseConnection())
+            using (var adb = SQLiteDatabaseConnectionBuilder.InMemory().BuildAsyncDatabaseConnection())
             {
                 var _0 = "hello";
                 var _1 = 1;
@@ -181,7 +249,10 @@ await adb.Query("Select ?, ?, ?", _0, _1, _2)
         [Fact]
         public void TestStatementCancellation()
         {
-            using (var adb = SQLite3.OpenInMemory().AsAsyncDatabaseConnection(TaskPoolScheduler.Default, 1))
+            var builder = SQLiteDatabaseConnectionBuilder.InMemory();
+            builder.ProgressHandlerInterval = 1;
+
+            using (var adb =  builder.BuildAsyncDatabaseConnection(TaskPoolScheduler.Default))
             {
                 var cts = new CancellationTokenSource();
                 Assert.ThrowsAsync<TaskCanceledException>(async () =>