Merge branch 'main' into chore/test-improvements

Author: LucDeCaf

PowerSync/PowerSync.Common/CHANGELOG.md

@@ -2,6 +2,49 @@
 
 ## 0.0.11-alpha.1
 
+- `StatusUpdated` and `StatusChanged` now both emit `SyncStatus` objects instead of just `StatusChanged`.
+- Converted most instances of a class inheriting from `EventStream<T>` into a class with an `EventManager` property called `Events`. This allows for subscribing to individual events instead of subscribing to all events and then filtering events manually.
+
+```csharp
+// Old
+var listener = db.Listen(cts.Token);
+foreach (PowerSyncDBEvent update in listener)
+{
+    // Manually filter updates
+    if (update.StatusChanged != null)
+    {
+        Console.WriteLine("status changed: " + update.StatusChanged!);
+    }
+}
+
+// Old (async)
+var listener = db.ListenAsync(cts.Token);
+await foreach (PowerSyncDBEvent update in listener)
+{
+    // Manually filter updates
+    if (update.StatusChanged != null)
+    {
+        Console.WriteLine("status changed: " + update.StatusChanged!);
+    }
+}
+
+// New
+var listener = db.Events.OnStatusChanged.Listen(cts.Token);
+await foreach (PowerSyncDBEvents.StatusChanged update in listener)
+{
+    // Events are filtered inherently
+    Console.WriteLine("status changed: " + update.Status);
+}
+
+// New (async) - recommended for most use cases
+var listener = db.Events.OnStatusChanged.ListenAsync(cts.Token);
+await foreach (PowerSyncDBEvents.StatusChanged update in listener)
+{
+    // Events are filtered inherently
+    Console.WriteLine("status changed: " + update.Status);
+}
+```
+
 - Pool read connections in `MDSQLiteAdapter`, improving performance in any case where multiple queries run simultaneously (eg. via `Watch`). The number of connections can be set via `MDSQLiteOptions.ReadPoolSize` and defaults to 5.
 - Updated to the latest version (0.4.11) of the core extension.
 - `MDSQLiteConnection` now runs query operations on another thread, which stops the caller thread from blocking.

PowerSync/PowerSync.Common/Client/ConnectionManager.cs

@@ -79,14 +79,25 @@ public class StoredConnectionOptions(
     public PowerSyncConnectionOptions Options { get; set; } = options;
 }
 
-public class ConnectionManagerEvent
+public class ConnectionManagerEvents : EventManager
 {
-    public StreamingSyncImplementation? SyncStreamCreated { get; set; }
+    public interface IConnectionManagerEvent;
+
+    public class SyncStreamCreatedEvent(StreamingSyncImplementation ssi) : IConnectionManagerEvent
+    {
+        public StreamingSyncImplementation SyncStreamCreated { get; set; } = ssi;
+    }
+
+    public EventStream<SyncStreamCreatedEvent> OnSyncStreamCreated { get; } = new();
+
+    public ConnectionManagerEvents()
+    {
+        Register(OnSyncStreamCreated);
+    }
 }
 
-public class ConnectionManager : EventStream<ConnectionManagerEvent>
+public class ConnectionManager : ICloseable
 {
-
     /// <summary>
     /// Tracks active connection attempts
     /// </summary>
@@ -122,6 +133,7 @@ public class ConnectionManager : EventStream<ConnectionManagerEvent>
 
     public IPowerSyncBackendConnector? Connector => PendingConnectionOptions?.Connector;
 
+    public ConnectionManagerEvents Events { get; protected set; } = new();
 
     public PowerSyncConnectionOptions? ConnectionOptions => PendingConnectionOptions?.Options;
 
@@ -148,9 +160,9 @@ public ConnectionManager(Func<IPowerSyncBackendConnector, CreateSyncImplementati
         SyncDisposer = null;
     }
 
-    public new void Close()
+    public void Close()
     {
-        base.Close();
+        Events.Close();
         SyncStreamImplementation?.Close();
         SyncDisposer?.Invoke();
     }
@@ -274,7 +286,7 @@ async Task InitSyncStream()
             RetryDelayMs = options.RetryDelayMs,
         });
 
-        Emit(new ConnectionManagerEvent { SyncStreamCreated = result.Sync });
+        Events.Emit(new ConnectionManagerEvents.SyncStreamCreatedEvent(result.Sync));
         SyncStreamImplementation = result.Sync;
         SyncDisposer = result.OnDispose;
         await SyncStreamImplementation.WaitForReady();

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -53,17 +53,45 @@ public class PowerSyncDatabaseOptions() : BasePowerSyncDatabaseOptions()
     public Func<IPowerSyncBackendConnector, Remote>? RemoteFactory { get; set; }
 }
 
-public class PowerSyncDBEvent : StreamingSyncImplementationEvent
+public class PowerSyncDBEvents : EventManager
 {
-    public bool? Initialized { get; set; }
-    public Schema? SchemaChanged { get; set; }
+    public interface IPowerSyncDBEvent;
 
-    public bool? Closing { get; set; }
+    public class InitializedEvent : IPowerSyncDBEvent;
+    public class ClosingEvent : IPowerSyncDBEvent;
+    public class ClosedEvent : IPowerSyncDBEvent;
+    public class SchemaChangedEvent(Schema schema) : IPowerSyncDBEvent
+    {
+        public Schema Schema { get; set; } = schema;
+    }
+    public class StatusChangedEvent(SyncStatus status) : IPowerSyncDBEvent
+    {
+        public SyncStatus Status { get; set; } = status;
+    }
+    public class StatusUpdatedEvent(SyncStatusOptions status) : IPowerSyncDBEvent
+    {
+        public SyncStatusOptions Status { get; set; } = status;
+    }
+
+    public EventStream<InitializedEvent> OnInitialized { get; } = new();
+    public EventStream<ClosingEvent> OnClosing { get; } = new();
+    public EventStream<ClosedEvent> OnClosed { get; } = new();
+    public EventStream<SchemaChangedEvent> OnSchemaChanged { get; } = new();
+    public EventStream<StatusChangedEvent> OnStatusChanged { get; } = new();
+    public EventStream<StatusUpdatedEvent> OnStatusUpdated { get; } = new();
 
-    public bool? Closed { get; set; }
+    public PowerSyncDBEvents()
+    {
+        Register(OnInitialized);
+        Register(OnClosing);
+        Register(OnClosed);
+        Register(OnSchemaChanged);
+        Register(OnStatusChanged);
+        Register(OnStatusUpdated);
+    }
 }
 
-public interface IPowerSyncDatabase : IEventStream<PowerSyncDBEvent>
+public interface IPowerSyncDatabase : ICloseableAsync
 {
     public Task Connect(IPowerSyncBackendConnector connector, PowerSyncConnectionOptions? options = null);
     public ISyncStream SyncStream(string name, Dictionary<string, object>? parameters = null);
@@ -96,20 +124,21 @@ public interface IPowerSyncDatabase : IEventStream<PowerSyncDBEvent>
 
     Task WriteTransaction(Func<ITransaction, Task> fn, DBLockOptions? options = null);
     Task<T> WriteTransaction<T>(Func<ITransaction, Task<T>> fn, DBLockOptions? options = null);
-
 }
 
-public class PowerSyncDatabase : EventStream<PowerSyncDBEvent>, IPowerSyncDatabase
+public class PowerSyncDatabase : IPowerSyncDatabase
 {
     public IDBAdapter Database { get; protected set; }
     private CompiledSchema schema;
 
     private static readonly int DEFAULT_WATCH_THROTTLE_MS = 30;
     private static readonly Regex POWERSYNC_TABLE_MATCH = new Regex(@"(^ps_data__|^ps_data_local__)", RegexOptions.Compiled);
 
-    public new bool Closed { get; protected set; }
+    public bool Closed { get; protected set; }
     public bool Ready { get; protected set; }
 
+    public PowerSyncDBEvents Events { get; protected set; } = new();
+
     protected Task IsReadyTask;
     protected ConnectionManager ConnectionManager;
 
@@ -138,7 +167,6 @@ public class PowerSyncDatabase : EventStream<PowerSyncDBEvent>, IPowerSyncDataba
     public bool Connected => CurrentStatus.Connected;
     public bool Connecting => CurrentStatus.Connecting;
 
-
     public PowerSyncConnectionOptions? ConnectionOptions => ConnectionManager.ConnectionOptions;
 
     public PowerSyncDatabase(PowerSyncDatabaseOptions options)
@@ -205,19 +233,16 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options)
                 });
 
                 var syncStreamStatusCts = CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token);
-                var syncStreamStatusListener = syncStreamImplementation.ListenAsync(syncStreamStatusCts.Token);
+                var syncStreamStatusListener = syncStreamImplementation.Events.OnStatusChanged.ListenAsync(syncStreamStatusCts.Token);
                 var _ = Task.Run(async () =>
                 {
                     await foreach (var update in syncStreamStatusListener)
                     {
-                        if (update.StatusChanged != null)
+                        CurrentStatus = new SyncStatus(new SyncStatusOptions(update.Status.Options)
                         {
-                            CurrentStatus = new SyncStatus(new SyncStatusOptions(update.StatusChanged.Options)
-                            {
-                                HasSynced = CurrentStatus?.HasSynced == true || update.StatusChanged.LastSyncedAt != null,
-                            });
-                            Emit(new PowerSyncDBEvent { StatusChanged = CurrentStatus });
-                        }
+                            HasSynced = CurrentStatus?.HasSynced == true || update.Status.LastSyncedAt != null,
+                        });
+                        Events.Emit(new PowerSyncDBEvents.StatusChangedEvent(CurrentStatus));
                     }
                 });
 
@@ -285,7 +310,7 @@ public async Task WaitForStatus(Func<SyncStatus, bool> predicate, CancellationTo
             ? CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token)
             : CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token, cancellationToken.Value);
 
-        var statusListener = ListenAsync(cts.Token);
+        var statusListener = Events.OnStatusChanged.ListenAsync(cts.Token);
 
         if (predicate(CurrentStatus))
         {
@@ -301,7 +326,7 @@ public async Task WaitForStatus(Func<SyncStatus, bool> predicate, CancellationTo
             {
                 await foreach (var update in statusListener)
                 {
-                    if ((update.StatusChanged != null) && predicate(update.StatusChanged))
+                    if (predicate(update.Status))
                     {
                         tcs.TrySetResult(true);
                         cts.Cancel();
@@ -322,7 +347,7 @@ protected async Task Initialize(PowerSyncDatabaseOptions options)
         await ResolveOfflineSyncStatus();
         await Database.Execute("PRAGMA RECURSIVE_TRIGGERS=TRUE");
         Ready = true;
-        Emit(new PowerSyncDBEvent { Initialized = true });
+        Events.Emit(new PowerSyncDBEvents.InitializedEvent());
     }
 
     private record VersionResult(string version);
@@ -366,7 +391,7 @@ protected async Task ResolveOfflineSyncStatus()
         if (!updatedStatus.IsEqual(CurrentStatus))
         {
             CurrentStatus = updatedStatus;
-            Emit(new PowerSyncDBEvent { StatusChanged = CurrentStatus });
+            Events.Emit(new PowerSyncDBEvents.StatusChangedEvent(CurrentStatus));
         }
     }
 
@@ -394,7 +419,7 @@ public async Task UpdateSchema(Schema schema)
         this.schema = compiledSchema;
         await Database.Execute("SELECT powersync_replace_schema(?)", [compiledSchema.ToJSON()]);
         await Database.RefreshSchema();
-        Emit(new PowerSyncDBEvent { SchemaChanged = schema });
+        Events.Emit(new PowerSyncDBEvents.SchemaChangedEvent(schema));
     }
 
     /// <summary>
@@ -458,7 +483,7 @@ await Database.WriteTransaction(async tx =>
 
         // The data has been deleted - reset the sync status
         CurrentStatus = new SyncStatus(new SyncStatusOptions());
-        Emit(new PowerSyncDBEvent { StatusChanged = CurrentStatus });
+        Events.Emit(new PowerSyncDBEvents.StatusChangedEvent(CurrentStatus));
     }
 
     /// <summary>
@@ -479,18 +504,16 @@ public ISyncStream SyncStream(string name, Dictionary<string, object>? parameter
     /// Once close is called, this connection cannot be used again - a new one
     /// must be constructed.
     /// </summary>
-    public new async Task Close()
+    public async Task Close()
     {
         await WaitForReady();
 
         if (Closed) return;
 
-        Emit(new PowerSyncDBEvent { Closing = true });
+        Events.Emit(new PowerSyncDBEvents.ClosingEvent());
 
         await Disconnect();
 
-        base.Close();
-
         masterCts.Cancel();
         masterCts.Dispose();
 
@@ -499,7 +522,10 @@ public ISyncStream SyncStream(string name, Dictionary<string, object>? parameter
 
         await Database.Close();
         Closed = true;
-        Emit(new PowerSyncDBEvent { Closed = true });
+
+        Events.Emit(new PowerSyncDBEvents.ClosedEvent());
+
+        Events.Close();
     }
 
     private record UploadQueueStatsSizeCountResult(long size, long count);
@@ -755,7 +781,7 @@ public IAsyncEnumerable<WatchOnChangeEvent> OnChange(SQLWatchOptions? options =
             ? CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token, options.Signal.Value)
             : CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token);
 
-        var listener = Database.ListenAsync(signal.Token);
+        var listener = Database.Events.OnTablesUpdated.ListenAsync(signal.Token);
 
         // Return the actual IAsyncEnumerable here, using OnChange as a synchronous wrapper that blocks until the
         // connection is established
@@ -764,7 +790,7 @@ public IAsyncEnumerable<WatchOnChangeEvent> OnChange(SQLWatchOptions? options =
 
     private async IAsyncEnumerable<WatchOnChangeEvent> OnChangeCore(
         HashSet<string> watchedTables,
-        IAsyncEnumerable<DBAdapterEvent> listener,
+        IAsyncEnumerable<DBAdapterEvents.TablesUpdatedEvent> listener,
         CancellationTokenSource signal,
         bool triggerImmediately
     )
@@ -818,7 +844,7 @@ public IAsyncEnumerable<T[]> Watch<T>(
         // so that table changes between Watch() being called and iteration starting are not missed.
         // This mirrors the pattern used in OnChange().
         var initialRestartCts = CancellationTokenSource.CreateLinkedTokenSource(signal.Token);
-        var initialListener = Database.ListenAsync(initialRestartCts.Token);
+        var initialListener = Database.Events.OnTablesUpdated.ListenAsync(initialRestartCts.Token);
 
         return WatchCore<T>(sql, parameters, options, signal, initialRestartCts, initialListener);
     }
@@ -829,22 +855,19 @@ private async IAsyncEnumerable<T[]> WatchCore<T>(
         SQLWatchOptions options,
         CancellationTokenSource signal,
         CancellationTokenSource initialRestartCts,
-        IAsyncEnumerable<DBAdapterEvent> initialListener
+        IAsyncEnumerable<DBAdapterEvents.TablesUpdatedEvent> initialListener
     )
     {
         var schemaChanged = new TaskCompletionSource<bool>();
 
         // Listen for schema changes in the background
         var schemaListenerTask = Task.Run(async () =>
         {
-            await foreach (var update in ListenAsync(signal.Token))
+            await foreach (var update in Events.OnSchemaChanged.ListenAsync(signal.Token))
             {
-                if (update.SchemaChanged != null)
-                {
-                    // Swap schemaChanged with an unresolved TCS
-                    var oldTcs = Interlocked.Exchange(ref schemaChanged, new());
-                    oldTcs.TrySetResult(true);
-                }
+                // Swap schemaChanged with an unresolved TCS
+                var oldTcs = Interlocked.Exchange(ref schemaChanged, new());
+                oldTcs.TrySetResult(true);
             }
         }, signal.Token);
 
@@ -897,7 +920,7 @@ IAsyncEnumerable<DBAdapterEvent> initialListener
                         // Establish a new listener BEFORE resolving source tables in the next iteration,
                         // so that changes during the async GetSourceTables call are not missed.
                         currentRestartCts = CancellationTokenSource.CreateLinkedTokenSource(signal.Token);
-                        currentListener = Database.ListenAsync(currentRestartCts.Token);
+                        currentListener = Database.Events.OnTablesUpdated.ListenAsync(currentRestartCts.Token);
                         oldRestartCts.Dispose();
 
                         break;
@@ -961,7 +984,7 @@ internal async Task<HashSet<string>> GetSourceTables(string sql, object?[]? para
 
     private async IAsyncEnumerable<WatchOnChangeEvent> OnRawTableChange(
         HashSet<string> watchedTables,
-        IAsyncEnumerable<DBAdapterEvent> listener,
+        IAsyncEnumerable<DBAdapterEvents.TablesUpdatedEvent> listener,
         [EnumeratorCancellation] CancellationToken token,
         bool triggerImmediately = false
     )
@@ -974,8 +997,7 @@ private async IAsyncEnumerable<WatchOnChangeEvent> OnRawTableChange(
         HashSet<string> changedTables = new();
         await foreach (var e in listener)
         {
-            if (e.TablesUpdated == null) continue;
-
+            // Extract the changed tables and intersect with the watched tables
             changedTables.Clear();
             GetTablesFromNotification(e.TablesUpdated, changedTables);
             changedTables.IntersectWith(watchedTables);