Native: Propagate stream subscriptions

Author: simolus3

packages/powersync_core/lib/src/database/native/native_powersync_database.dart

@@ -142,6 +142,7 @@ class PowerSyncDatabaseImpl
 
     bool triedSpawningIsolate = false;
     StreamSubscription<UpdateNotification>? crudUpdateSubscription;
+    StreamSubscription<void>? activeStreamsSubscription;
     final receiveMessages = ReceivePort();
     final receiveUnhandledErrors = ReceivePort();
     final receiveExit = ReceivePort();
@@ -159,6 +160,7 @@ class PowerSyncDatabaseImpl
 
       // Cleanup
       crudUpdateSubscription?.cancel();
+      activeStreamsSubscription?.cancel();
       receiveMessages.close();
       receiveUnhandledErrors.close();
       receiveExit.close();
@@ -200,6 +202,10 @@ class PowerSyncDatabaseImpl
           crudUpdateSubscription = crudStream.listen((event) {
             port.send(['update']);
           });
+
+          activeStreamsSubscription = activeStreams.listen((streams) {
+            port.send(['changed_subscriptions', streams]);
+          });
         } else if (action == 'uploadCrud') {
           await (data[1] as PortCompleter).handle(() async {
             await connector.uploadData(this);
@@ -368,6 +374,9 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
         }
       } else if (action == 'close') {
         await shutdown();
+      } else if (action == 'changed_subscriptions') {
+        openedStreamingSync
+            ?.updateSubscriptions(action[1] as List<SubscribedStream>);
       }
     }
   });

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -38,7 +38,7 @@ class StreamingSyncImplementation implements StreamingSync {
   final BucketStorage adapter;
   final InternalConnector connector;
   final ResolvedSyncOptions options;
-  final List<SubscribedStream> activeSubscriptions;
+  List<SubscribedStream> _activeSubscriptions;
 
   final Logger logger;
 
@@ -72,7 +72,7 @@ class StreamingSyncImplementation implements StreamingSync {
     required this.crudUpdateTriggerStream,
     required this.options,
     required http.Client client,
-    this.activeSubscriptions = const [],
+    List<SubscribedStream> activeSubscriptions = const [],
     Mutex? syncMutex,
     Mutex? crudMutex,
     Logger? logger,
@@ -84,7 +84,8 @@ class StreamingSyncImplementation implements StreamingSync {
         syncMutex = syncMutex ?? Mutex(identifier: "sync-$identifier"),
         crudMutex = crudMutex ?? Mutex(identifier: "crud-$identifier"),
         _userAgentHeaders = userAgentHeaders(),
-        logger = logger ?? isolateLogger;
+        logger = logger ?? isolateLogger,
+        _activeSubscriptions = activeSubscriptions;
 
   Duration get _retryDelay => options.retryDelay;
 
@@ -128,6 +129,13 @@ class StreamingSyncImplementation implements StreamingSync {
     return _abort?.aborted ?? false;
   }
 
+  void updateSubscriptions(List<SubscribedStream> streams) {
+    _activeSubscriptions = streams;
+    if (_nonLineSyncEvents.hasListener) {
+      _nonLineSyncEvents.add(const AbortCurrentIteration());
+    }
+  }
+
   @override
   Future<void> streamingSync() async {
     try {
@@ -609,7 +617,7 @@ final class _ActiveRustStreamingIteration {
           'parameters': sync.options.params,
           'schema': convert.json.decode(sync.schemaJson),
           'include_defaults': sync.options.includeDefaultStreams,
-          'active_streams': sync.activeSubscriptions
+          'active_streams': sync._activeSubscriptions
               .map((s) => {'name': s.name, 'params': s.parameters})
         }),
       );