Sync streams

demos/supabase-todolist/lib/widgets/todo_list_page.dart

@@ -1,4 +1,5 @@
 import 'package:flutter/material.dart';
+import 'package:powersync/powersync.dart';
 
 import '../powersync.dart';
 import './status_app_bar.dart';
@@ -38,30 +39,82 @@ class TodoListPage extends StatelessWidget {
   }
 }
 
-class TodoListWidget extends StatelessWidget {
+class TodoListWidget extends StatefulWidget {
   final TodoList list;
 
   const TodoListWidget({super.key, required this.list});
 
+  @override
+  State<TodoListWidget> createState() => _TodoListWidgetState();
+}
+
+class _TodoListWidgetState extends State<TodoListWidget> {
+  SyncStreamSubscription? _listSubscription;
+
+  void _subscribe(String listId) {
+    db
+        .syncStream('todos', {'list': listId})
+        .subscribe(ttl: const Duration(hours: 1))
+        .then((sub) {
+          if (mounted && widget.list.id == listId) {
+            setState(() {
+              _listSubscription = sub;
+            });
+          } else {
+            sub.unsubscribe();
+          }
+        });
+  }
+
+  @override
+  void initState() {
+    super.initState();
+    _subscribe(widget.list.id);
+  }
+
+  @override
+  void didUpdateWidget(covariant TodoListWidget oldWidget) {
+    super.didUpdateWidget(oldWidget);
+    _subscribe(widget.list.id);
+  }
+
+  @override
+  void dispose() {
+    super.dispose();
+    _listSubscription?.unsubscribe();
+  }
+
   @override
   Widget build(BuildContext context) {
     return StreamBuilder(
-      stream: TodoList.watchSyncStatus().map((e) => e.hasSynced),
-      initialData: db.currentStatus.hasSynced,
+      stream: db.statusStream,
+      initialData: db.currentStatus,
       builder: (context, snapshot) {
-        return StreamBuilder(
-          stream: list.watchItems(),
-          builder: (context, snapshot) {
-            final items = snapshot.data ?? const [];
-
-            return ListView(
-              padding: const EdgeInsets.symmetric(vertical: 8.0),
-              children: items.map((todo) {
-                return TodoItemWidget(todo: todo);
-              }).toList(),
-            );
-          },
-        );
+        final hasSynced = switch (_listSubscription) {
+              null => null,
+              final sub => snapshot.requireData.statusFor(sub),
+            }
+                ?.subscription
+                .hasSynced ??
+            false;
+
+        if (!hasSynced) {
+          return const CircularProgressIndicator();
+        } else {
+          return StreamBuilder(
+            stream: widget.list.watchItems(),
+            builder: (context, snapshot) {
+              final items = snapshot.data ?? const [];
+
+              return ListView(
+                padding: const EdgeInsets.symmetric(vertical: 8.0),
+                children: items.map((todo) {
+                  return TodoItemWidget(todo: todo);
+                }).toList(),
+              );
+            },
+          );
+        }
       },
     );
   }

demos/supabase-todolist/pubspec.lock

@@ -605,10 +605,10 @@ packages:
     dependency: transitive
     description:
       name: sqlite3
-      sha256: "310af39c40dd0bb2058538333c9d9840a2725ae0b9f77e4fd09ad6696aa8f66e"
+      sha256: f393d92c71bdcc118d6203d07c991b9be0f84b1a6f89dd4f7eed348131329924
       url: "https://pub.dev"
     source: hosted
-    version: "2.7.5"
+    version: "2.9.0"
   sqlite3_flutter_libs:
     dependency: transitive
     description:
@@ -621,18 +621,17 @@ packages:
     dependency: transitive
     description:
       name: sqlite3_web
-      sha256: "967e076442f7e1233bd7241ca61f3efe4c7fc168dac0f38411bdb3bdf471eb3c"
+      sha256: "0f6ebcb4992d1892ac5c8b5ecd22a458ab9c5eb6428b11ae5ecb5d63545844da"
       url: "https://pub.dev"
     source: hosted
-    version: "0.3.1"
+    version: "0.3.2"
   sqlite_async:
     dependency: "direct main"
     description:
-      name: sqlite_async
-      sha256: "9332aedd311a19dd215dcb55729bc68dc587dc7655b569ab8819b68ee0be0082"
-      url: "https://pub.dev"
-    source: hosted
-    version: "0.11.7"
+      path: "/Users/simon/src/sqlite_async.dart/packages/sqlite_async"
+      relative: false
+    source: path
+    version: "0.12.1"
   stack_trace:
     dependency: transitive
     description:

packages/powersync_core/lib/powersync_core.dart

@@ -11,6 +11,7 @@ export 'src/log.dart';
 export 'src/open_factory.dart';
 export 'src/schema.dart';
 export 'src/sync/options.dart' hide ResolvedSyncOptions;
+export 'src/sync/stream.dart' hide CoreActiveStreamSubscription;
 export 'src/sync/sync_status.dart'
-    hide BucketProgress, InternalSyncDownloadProgress;
+    hide BucketProgress, InternalSyncDownloadProgress, InternalSyncStatusAccess;
 export 'src/uuid.dart';

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

@@ -133,13 +133,16 @@ class PowerSyncDatabaseImpl
   Future<void> connectInternal({
     required PowerSyncBackendConnector connector,
     required ResolvedSyncOptions options,
+    required List<SubscribedStream> initiallyActiveStreams,
+    required Stream<List<({String name, String parameters})>> activeStreams,
     required AbortController abort,
     required Zone asyncWorkZone,
   }) async {
     final dbRef = database.isolateConnectionFactory();
 
     bool triedSpawningIsolate = false;
     StreamSubscription<UpdateNotification>? crudUpdateSubscription;
+    StreamSubscription<void>? activeStreamsSubscription;
     final receiveMessages = ReceivePort();
     final receiveUnhandledErrors = ReceivePort();
     final receiveExit = ReceivePort();
@@ -157,6 +160,7 @@ class PowerSyncDatabaseImpl
 
       // Cleanup
       crudUpdateSubscription?.cancel();
+      activeStreamsSubscription?.cancel();
       receiveMessages.close();
       receiveUnhandledErrors.close();
       receiveExit.close();
@@ -198,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);
@@ -366,6 +374,9 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
         }
       } else if (action == 'close') {
         await shutdown();
+      } else if (action == 'changed_subscriptions') {
+        openedStreamingSync
+            ?.updateSubscriptions(message[1] as List<SubscribedStream>);
       }
     }
   });

packages/powersync_core/lib/src/database/powersync_database_impl_stub.dart

@@ -7,6 +7,7 @@ import 'package:powersync_core/src/abort_controller.dart';
 import 'package:powersync_core/src/database/powersync_db_mixin.dart';
 import 'package:powersync_core/src/open_factory/abstract_powersync_open_factory.dart';
 import '../sync/options.dart';
+import '../sync/streaming_sync.dart';
 import 'powersync_database.dart';
 
 import '../connector.dart';
@@ -115,6 +116,8 @@ class PowerSyncDatabaseImpl
   Future<void> connectInternal({
     required PowerSyncBackendConnector connector,
     required AbortController abort,
+    required List<SubscribedStream> initiallyActiveStreams,
+    required Stream<List<SubscribedStream>> activeStreams,
     required Zone asyncWorkZone,
     required ResolvedSyncOptions options,
   }) {