Using event stream to interface with central control call.

Author: Chriztiaan

PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs

@@ -396,6 +396,12 @@ protected record StreamingSyncIterationResult
         public bool? ImmediateRestart { get; init; }
     }
 
+    protected record EnqueuedCommand
+    {
+        public string Command { get; init; } = null!;
+        public object? Payload { get; init; }
+    }
+
 
     protected async Task<StreamingSyncIterationResult> StreamingSyncIteration(CancellationToken signal, PowerSyncConnectionOptions? options)
     {
@@ -423,12 +429,14 @@ protected async Task<StreamingSyncIterationResult> StreamingSyncIteration(Cancel
         });
     }
 
+
     protected async Task<StreamingSyncIterationResult> RustStreamingSyncIteration(CancellationToken? signal, RequiredPowerSyncConnectionOptions resolvedOptions)
     {
         Task? receivingLines = null;
         bool hadSyncLine = false;
         bool hideDisconnectOnRestart = false;
 
+        var controlInvocations = (EventStream<EnqueuedCommand>)null!;
 
         var nestedCts = new CancellationTokenSource();
         signal?.Register(() => { nestedCts.Cancel(); });
@@ -442,32 +450,52 @@ async Task Connect(EstablishSyncStream instruction)
                 Data = instruction.Request
             };
 
-            var stream = await Options.Remote.PostStreamRaw(syncOptions);
-            using var reader = new StreamReader(stream, Encoding.UTF8);
-
-            syncOptions.CancellationToken.Register(() =>
+            controlInvocations = new EventStream<EnqueuedCommand>();
+            try
             {
-                try { stream?.Close(); } catch { }
-            });
+                controlInvocations?.RunListenerAsync(async (line) =>
+                {
+                    await Control(line.Command, line.Payload);
 
-            UpdateSyncStatus(new SyncStatusOptions
-            {
-                Connected = true
-            });
+                    // Triggers a local CRUD upload when the first sync line has been received.
+                    // This allows uploading local changes that have been made while offline or disconnected.
+                    if (!hadSyncLine)
+                    {
+                        TriggerCrudUpload();
+                        hadSyncLine = true;
+                    }
+                });
 
-            string? line;
-            while ((line = await reader.ReadLineAsync()) != null)
-            {
-                await Control(PowerSyncControlCommand.PROCESS_TEXT_LINE, line);
+                var stream = await Options.Remote.PostStreamRaw(syncOptions);
+                using var reader = new StreamReader(stream, Encoding.UTF8);
 
-                // Triggers a local CRUD upload when the first sync line has been received.
-                // This allows uploading local changes that have been made while offline or disconnected.
-                if (!hadSyncLine)
+                syncOptions.CancellationToken.Register(() =>
                 {
-                    TriggerCrudUpload();
-                    hadSyncLine = true;
+                    try { stream?.Close(); } catch { }
+                });
+
+                UpdateSyncStatus(new SyncStatusOptions
+                {
+                    Connected = true
+                });
+
+                string? line;
+                while ((line = await reader.ReadLineAsync()) != null)
+                {
+                    controlInvocations?.Emit(new EnqueuedCommand
+                    {
+                        Command = PowerSyncControlCommand.PROCESS_TEXT_LINE,
+                        Payload = line
+                    });
+
                 }
             }
+            finally
+            {
+                var activeInstructions = controlInvocations;
+                controlInvocations = null;
+                activeInstructions?.Close();
+            }
         }
 
         async Task Stop()
@@ -553,7 +581,10 @@ async Task HandleInstruction(Instruction instruction)
                         try
                         {
                             await Options.Remote.FetchCredentials();
-                            await Control(PowerSyncControlCommand.NOTIFY_TOKEN_REFRESHED);
+                            controlInvocations?.Emit(new EnqueuedCommand
+                            {
+                                Command = PowerSyncControlCommand.NOTIFY_TOKEN_REFRESHED
+                            });
                         }
                         catch (Exception err)
                         {
@@ -586,9 +617,12 @@ async Task HandleInstruction(Instruction instruction)
             {
                 Task.Run(async () =>
                 {
-                    if (!nestedCts.IsCancellationRequested)
+                    if (controlInvocations != null && !controlInvocations.Closed)
                     {
-                        await Control(PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED);
+                        controlInvocations?.Emit(new EnqueuedCommand
+                        {
+                            Command = PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED
+                        });
                     }
                 });
             };

PowerSync/PowerSync.Common/Utils/EventStream.cs

@@ -25,6 +25,8 @@ CancellationTokenSource RunListenerAsync(
 public class EventStream<T> : IEventStream<T>
 {
 
+    public bool Closed = false;
+
     // Closest implementation to a ConcurrentSet<T> in .Net
     private readonly ConcurrentDictionary<Channel<T>, byte> subscribers = new();
 
@@ -158,6 +160,7 @@ public void Close()
             subscriber.Writer.TryComplete();
             RemoveSubscriber(subscriber);
         }
+        Closed = true;
     }
 
     private void RemoveSubscriber(Channel<T> channel)