Add Deregister method + tests

Author: LucDeCaf

PowerSync/PowerSync.Common/Utils/EventManager.cs

@@ -7,6 +7,12 @@ public interface IEventManager : ICloseable
     /// </summary>
     void Register<T>(EventStream<T> stream);
 
+    /// <summary>
+    /// <para>Deregisters the stream associated with the EventManager.</para>
+    /// <para>This does NOT close the stream in the default implementation.</para>
+    /// </summary>
+    bool Deregister<T>();
+
     /// <summary>
     /// Attempts to retreive the EventStream associated with events of type T.
     /// </summary>
@@ -32,6 +38,11 @@ public void Register<T>(EventStream<T> stream)
         _streams[typeof(T)] = stream;
     }
 
+    public bool Deregister<T>()
+    {
+        return _streams.Remove(typeof(T));
+    }
+
     public bool TryGetStream<T>(out EventStream<T> stream)
     {
         if (_streams.TryGetValue(typeof(T), out var streamObj))

Tests/PowerSync/PowerSync.Common.Tests/EventStreamTests.cs

@@ -5,7 +5,6 @@ namespace PowerSync.Common.Tests;
 
 public class EventStreamTests
 {
-
     [Fact]
     public async Task EventStream_ShouldReceiveTwoMessages_Async()
     {
@@ -109,4 +108,136 @@ public async Task EventStream_ShouldReceiveTwoMessages_Sync()
         Assert.Contains(status2, receivedMessages);
         Assert.Equal(0, eventStream.SubscriberCount());
     }
+
+    [Fact]
+    public async Task EventManager_RegistersStreamsCorrectly()
+    {
+        var manager = new EventManager();
+        var stream1 = new EventStream<bool>();
+        var stream2 = new EventStream<string>();
+        var stream3 = new EventStream<int>(); // Control
+
+        manager.Register(stream1);
+        manager.Register(stream2);
+
+        Assert.True(manager.TryGetStream<bool>(out var obtainedStream1));
+        Assert.True(manager.TryGetStream<string>(out var obtainedStream2));
+        Assert.False(manager.TryGetStream<int>(out var obtainedStream3));
+
+        Assert.Equal(stream1, obtainedStream1);
+        Assert.Equal(stream2, obtainedStream2);
+        Assert.Equal(null, obtainedStream3);
+
+        manager.Close();
+    }
+
+    [Fact]
+    public async Task EventManager_CloseRemovesAndClosesStreams()
+    {
+        var manager = new EventManager();
+        var stream1 = new EventStream<bool>();
+        var stream2 = new EventStream<string>();
+        var stream3 = new EventStream<int>(); // Control
+
+        manager.Register(stream1);
+        manager.Register(stream2);
+
+        manager.Close();
+
+        Assert.False(manager.TryGetStream<bool>(out var obtainedStream1));
+        Assert.False(manager.TryGetStream<string>(out var obtainedStream2));
+        Assert.False(manager.TryGetStream<int>(out var obtainedStream3));
+
+        Assert.Equal(null, obtainedStream1);
+        Assert.Equal(null, obtainedStream2);
+        Assert.Equal(null, obtainedStream3);
+
+        Assert.True(stream1.Closed);
+        Assert.True(stream2.Closed);
+        Assert.False(stream3.Closed);
+    }
+
+    [Fact(Timeout = 2000)]
+    public async Task EventManager_ShouldReceiveEmittedEvents()
+    {
+        var manager = new EventManager();
+        var stream1 = new EventStream<bool>();
+        var stream2 = new EventStream<string>();
+
+        manager.Register(stream1);
+        manager.Register(stream2);
+
+        var cts1 = new CancellationTokenSource();
+        var listener1 = stream1.ListenAsync(cts1.Token);
+        Assert.True(manager.TryEmit(false));
+        Assert.True(manager.TryEmit(false));
+        Assert.True(manager.TryEmit(true));
+        int eventCount = 0;
+
+        await foreach (var evt in listener1)
+        {
+            eventCount++;
+            if (evt == true)
+            {
+                cts1.Cancel();
+            }
+        }
+
+        Assert.Equal(3, eventCount);
+
+        var cts2 = new CancellationTokenSource();
+        var listener2 = stream2.ListenAsync(cts2.Token);
+        Assert.True(manager.TryEmit("hi"));
+        Assert.True(manager.TryEmit("hello"));
+        Assert.True(manager.TryEmit("sup"));
+        Assert.True(manager.TryEmit("STOP"));
+        eventCount = 0;
+
+        await foreach (var evt in listener2)
+        {
+            eventCount++;
+            if (evt == "STOP")
+            {
+                cts2.Cancel();
+            }
+        }
+
+        Assert.Equal(4, eventCount);
+
+        manager.Close();
+    }
+
+    [Fact]
+    public async Task EventManager_ShouldNotReceiveEventsAfterDeregistering()
+    {
+        var manager = new EventManager();
+        var stream = new EventStream<string>();
+
+        manager.Register(stream);
+
+        var cts = new CancellationTokenSource();
+        var listener = stream.ListenAsync(cts.Token);
+        var sem = new SemaphoreSlim(0);
+        int eventCount = 0;
+
+        _ = Task.Run(async () =>
+        {
+            sem.Release();
+            await foreach (var evt in listener)
+            {
+                eventCount++;
+                sem.Release();
+            }
+        }, cts.Token);
+        Assert.True(await sem.WaitAsync(100));
+
+        Assert.True(manager.Deregister<string>());
+
+        Assert.False(manager.TryEmit("invalid"));
+        Assert.False(await sem.WaitAsync(100));
+
+        // Cleanup
+        cts.Cancel();
+        manager.Close();
+    }
 }