Improve API for registering streams

Author: LucDeCaf

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -53,7 +53,7 @@ public class PowerSyncDatabaseOptions() : BasePowerSyncDatabaseOptions()
     public Func<IPowerSyncBackendConnector, Remote>? RemoteFactory { get; set; }
 }
 
-public class PowerSyncDBEvents : EventManager<PowerSyncDBEvents.IPowerSyncDBEvent>
+public class PowerSyncDBEvents : EventManager
 {
     public interface IPowerSyncDBEvent;
 
@@ -80,60 +80,18 @@ public class StatusUpdatedEvent(SyncStatusOptions status) : IPowerSyncDBEvent
     public EventStream<StatusChangedEvent> OnStatusChanged { get; } = new();
     public EventStream<StatusUpdatedEvent> OnStatusUpdated { get; } = new();
 
-    public override bool TryGetStream<T>(out EventStream<T> stream)
+    public PowerSyncDBEvents()
     {
-        if (typeof(T) == typeof(PowerSyncDBEvents.InitializedEvent))
-        {
-            stream = (EventStream<T>)(object)OnInitialized;
-            return true;
-        }
-
-        if (typeof(T) == typeof(PowerSyncDBEvents.ClosingEvent))
-        {
-            stream = (EventStream<T>)(object)OnClosing;
-            return true;
-        }
-
-        if (typeof(T) == typeof(PowerSyncDBEvents.ClosedEvent))
-        {
-            stream = (EventStream<T>)(object)OnClosed;
-            return true;
-        }
-
-        if (typeof(T) == typeof(PowerSyncDBEvents.SchemaChangedEvent))
-        {
-            stream = (EventStream<T>)(object)OnSchemaChanged;
-            return true;
-        }
-
-        if (typeof(T) == typeof(PowerSyncDBEvents.StatusChangedEvent))
-        {
-            stream = (EventStream<T>)(object)OnStatusChanged;
-            return true;
-        }
-
-        if (typeof(T) == typeof(PowerSyncDBEvents.StatusUpdatedEvent))
-        {
-            stream = (EventStream<T>)(object)OnStatusUpdated;
-            return true;
-        }
-
-        stream = null!;
-        return false;
-    }
-
-    public override void Close()
-    {
-        OnInitialized.Close();
-        OnClosing.Close();
-        OnClosed.Close();
-        OnSchemaChanged.Close();
-        OnStatusChanged.Close();
-        OnStatusUpdated.Close();
+        Register(OnInitialized);
+        Register(OnClosing);
+        Register(OnClosed);
+        Register(OnSchemaChanged);
+        Register(OnStatusChanged);
+        Register(OnStatusUpdated);
     }
 }
 
-public interface IPowerSyncDatabase
+public interface IPowerSyncDatabase : ICloseableAsync
 {
     public Task Connect(IPowerSyncBackendConnector connector, PowerSyncConnectionOptions? options = null);
     public ISyncStream SyncStream(string name, Dictionary<string, object>? parameters = null);

PowerSync/PowerSync.Common/Utils/EventManager.cs

@@ -0,0 +1,80 @@
+namespace PowerSync.Common.Utils;
+
+public interface IEventManager : ICloseable
+{
+    /// <summary>
+    /// Registers a new EventStream into the EventManager.
+    /// </summary>
+    void Register<T>(EventStream<T> stream);
+
+    /// <summary>
+    /// Attempts to retreive the EventStream associated with events of type T.
+    /// </summary>
+    bool TryGetStream<T>(out EventStream<T> stream);
+
+    /// <summary>
+    /// Posts a message to the stream managing events of type T.
+    /// </summary>
+    void Emit<T>(T evt);
+
+    /// <summary>
+    /// Attemps to post a message to the stream managing events of type T.
+    /// </summary>
+    bool TryEmit<T>(T evt);
+}
+
+public class EventManager : IEventManager
+{
+    private readonly Dictionary<Type, object> _streams = new();
+
+    public void Register<T>(EventStream<T> stream)
+    {
+        _streams[typeof(T)] = stream;
+    }
+
+    public bool TryGetStream<T>(out EventStream<T> stream)
+    {
+        if (_streams.TryGetValue(typeof(T), out var streamObj))
+        {
+            stream = (EventStream<T>)streamObj;
+            return true;
+        }
+        stream = null!;
+        return false;
+    }
+
+    public void Emit<T>(T evt)
+    {
+        if (TryGetStream<T>(out var stream))
+        {
+            stream.Emit(evt);
+        }
+        else
+        {
+            throw new InvalidOperationException($"No stream emits events of type {nameof(T)}.");
+        }
+    }
+
+    public bool TryEmit<T>(T evt)
+    {
+        if (TryGetStream<T>(out var stream))
+        {
+            stream.Emit(evt);
+            return true;
+        }
+        else
+        {
+            return false;
+        }
+    }
+
+    public void Close()
+    {
+        foreach (var kvp in _streams)
+        {
+            ((ICloseable)kvp.Value).Close();
+        }
+        _streams.Clear();
+    }
+}
+

PowerSync/PowerSync.Common/Utils/EventStream.cs

@@ -4,7 +4,7 @@ namespace PowerSync.Common.Utils;
 using System.Runtime.CompilerServices;
 using System.Threading.Channels;
 
-public interface IEventStream<T>
+public interface IEventStream<T> : ICloseable
 {
     void Emit(T item);
 
@@ -13,8 +13,6 @@ public interface IEventStream<T>
     IEnumerable<T> Listen(CancellationToken cancellationToken);
 
     IAsyncEnumerable<T> ListenAsync(CancellationToken cancellationToken);
-
-    void Close();
 }
 
 public class EventStream<T> : IEventStream<T>

PowerSync/PowerSync.Common/Utils/ICloseable.cs

@@ -0,0 +1,12 @@
+namespace PowerSync.Common.Utils;
+
+public interface ICloseable
+{
+    public void Close();
+}
+
+public interface ICloseableAsync
+{
+    public Task Close();
+}
+