Add EventManager class (#48)

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();