Use new pattern around codebase

Author: LucDeCaf

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

@@ -233,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,
-                            });
-                            Events.Emit(new PowerSyncDBEvents.StatusChangedEvent(CurrentStatus));
-                        }
+                            HasSynced = CurrentStatus?.HasSynced == true || update.Status.LastSyncedAt != null,
+                        });
+                        Events.Emit(new PowerSyncDBEvents.StatusChangedEvent(CurrentStatus));
                     }
                 });
 
@@ -784,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
@@ -793,7 +790,7 @@ public IAsyncEnumerable<WatchOnChangeEvent> OnChange(SQLWatchOptions? options =
 
     private async IAsyncEnumerable<WatchOnChangeEvent> OnChangeCore(
         HashSet<string> watchedTables,
-        IAsyncEnumerable<DBAdapterEvent> listener,
+        IAsyncEnumerable<DBAdapterEvents.TablesUpdatedEvent> listener,
         [EnumeratorCancellation] CancellationToken signal,
         bool triggerImmediately
     )
@@ -807,7 +804,6 @@ bool triggerImmediately
         await foreach (var e in listener)
         {
             if (signal.IsCancellationRequested) yield break;
-            if (e.TablesUpdated == null) continue;
 
             changedTables.Clear();
             GetTablesFromNotification(e.TablesUpdated, changedTables);
@@ -857,7 +853,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);
     }
@@ -868,7 +864,7 @@ private async IAsyncEnumerable<T[]> WatchCore<T>(
         SQLWatchOptions options,
         CancellationTokenSource signal,
         CancellationTokenSource initialRestartCts,
-        IAsyncEnumerable<DBAdapterEvent> initialListener
+        IAsyncEnumerable<DBAdapterEvents.TablesUpdatedEvent> initialListener
     )
     {
         var schemaChanged = new TaskCompletionSource<bool>();
@@ -930,7 +926,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);
 
                     break;
                 }
@@ -977,7 +973,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
     )
@@ -990,19 +986,16 @@ private async IAsyncEnumerable<WatchOnChangeEvent> OnRawTableChange(
         HashSet<string> changedTables = new();
         await foreach (var e in listener)
         {
-            if (e.TablesUpdated != null)
-            {
-                if (token.IsCancellationRequested) break;
+            if (token.IsCancellationRequested) break;
 
-                // Extract the changed tables and intersect with the watched tables
-                changedTables.Clear();
-                GetTablesFromNotification(e.TablesUpdated, changedTables);
-                changedTables.IntersectWith(watchedTables);
+            // Extract the changed tables and intersect with the watched tables
+            changedTables.Clear();
+            GetTablesFromNotification(e.TablesUpdated, changedTables);
+            changedTables.IntersectWith(watchedTables);
 
-                if (changedTables.Count == 0) continue;
+            if (changedTables.Count == 0) continue;
 
-                yield return new WatchOnChangeEvent { ChangedTables = [.. changedTables] };
-            }
+            yield return new WatchOnChangeEvent { ChangedTables = [.. changedTables] };
         }
     }
 

PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs

@@ -100,13 +100,23 @@ public static class PSInternalTable
     public static readonly string UNTYPED = "ps_untyped";
 }
 
-public class BucketStorageEvent
+public class BucketStorageEvents : EventManager
 {
-    public bool CrudUpdate { get; set; }
+    public interface IBucketStorageEvent;
+
+    public class CrudUpdateEvent : IBucketStorageEvent;
+
+    public EventStream<CrudUpdateEvent> OnCrudUpdate = new();
+
+    public BucketStorageEvents()
+    {
+        Register(OnCrudUpdate);
+    }
 }
 
-public interface IBucketStorageAdapter : IEventStream<BucketStorageEvent>
+public interface IBucketStorageAdapter : ICloseable
 {
+    BucketStorageEvents Events { get; }
     Task Init();
 
     Task<CrudEntry?> NextCrudItem();

PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs

@@ -12,13 +12,13 @@ namespace PowerSync.Common.Client.Sync.Bucket;
 
 using PowerSync.Common.DB;
 using PowerSync.Common.DB.Crud;
-using PowerSync.Common.Utils;
 
-public class SqliteBucketStorage : EventStream<BucketStorageEvent>, IBucketStorageAdapter
+public class SqliteBucketStorage : IBucketStorageAdapter
 {
-
     public static readonly string MAX_OP_ID = "9223372036854775807";
 
+    public BucketStorageEvents Events { get; } = new();
+
     private readonly IDBAdapter db;
     private bool hasCompletedSync;
     private readonly HashSet<string> tableNames;
@@ -27,29 +27,27 @@ public class SqliteBucketStorage : EventStream<BucketStorageEvent>, IBucketStora
     private readonly ILogger logger;
 
     private readonly CancellationTokenSource updateCts;
+    private readonly Task updateTask;
 
     private record ExistingTableRowsResult(string name);
 
     public SqliteBucketStorage(IDBAdapter db, ILogger? logger = null)
     {
         this.db = db;
-        this.logger = logger ?? NullLogger.Instance; ;
+        this.logger = logger ?? NullLogger.Instance;
         hasCompletedSync = false;
         tableNames = [];
 
         updateCts = new CancellationTokenSource();
 
-        var _ = Task.Run(() =>
+        updateTask = Task.Run(() =>
         {
-            foreach (var update in db.Listen(updateCts.Token))
+            foreach (var update in db.Events.OnTablesUpdated.Listen(updateCts.Token))
             {
-                if (update.TablesUpdated != null)
+                var tables = DBAdapterUtils.ExtractTableUpdates(update.TablesUpdated);
+                if (tables.Contains(PSInternalTable.CRUD))
                 {
-                    var tables = DBAdapterUtils.ExtractTableUpdates(update.TablesUpdated);
-                    if (tables.Contains(PSInternalTable.CRUD))
-                    {
-                        Emit(new BucketStorageEvent { CrudUpdate = true });
-                    }
+                    Events.Emit(new BucketStorageEvents.CrudUpdateEvent());
                 }
             }
         });
@@ -67,10 +65,11 @@ public async Task Init()
         }
     }
 
-    public new void Close()
+    public void Close()
     {
         updateCts.Cancel();
-        base.Close();
+        try { updateTask.Wait(2000); } catch (Exception) { }
+        Events.Close();
     }
 
     private record ClientIdResult(string? client_id);