Flexible sync subscribe/unsubscribe API

CHANGELOG.md

@@ -4,7 +4,8 @@
 * Support ReamSet.freeze() ([#1342](https://github.com/realm/realm-dart/pull/1342))
 * Added support for query on `RealmSet`. ([#1346](https://github.com/realm/realm-dart/pull/1346))
 * Support for passing `List`, `Set` or `Iterable` arguments to queries with `IN`-operators. ([#1346](https://github.com/realm/realm-dart/pull/1346))
-
+* Added new flexible sync API `RealmResults.subscribe()` and `RealmResults.unsubscribe()` as an easy way to create subscriptions and download data in background. Added named parameter to `MutableSubscriptionSet.clear({bool unnamedOnly = false})` for removing all the unnamed subscriptions. ([#1354](https://github.com/realm/realm-dart/pull/1354))
+* Added `cancellationToken` parameter to `Session.waitForDownload()`, `Session.waitForUpload()` and `SubscriptionSet.waitForSynchronization()`. ([#1354](https://github.com/realm/realm-dart/pull/1354))
 
 ### Fixed
 * Fixed an early unlock race condition during client reset callbacks. ([#1335](https://github.com/realm/realm-dart/pull/1335))

lib/src/native/realm_core.dart

@@ -413,17 +413,22 @@ class _RealmCore {
   }
 
   static void _stateChangeCallback(Object userdata, int state) {
-    final completer = userdata as Completer<SubscriptionSetState>;
-
+    final completer = userdata as CancellableCompleter<SubscriptionSetState>;
+    if (completer.isCompleted || completer.isCancelled) {
+      return;
+    }
     completer.complete(SubscriptionSetState.values[state]);
   }
 
-  Future<SubscriptionSetState> waitForSubscriptionSetStateChange(SubscriptionSet subscriptions, SubscriptionSetState notifyWhen) {
-    final completer = Completer<SubscriptionSetState>();
-    final callback = Pointer.fromFunction<Void Function(Handle, Int32)>(_stateChangeCallback);
-    final userdata = _realmLib.realm_dart_userdata_async_new(completer, callback.cast(), scheduler.handle._pointer);
-    _realmLib.realm_sync_on_subscription_set_state_change_async(subscriptions.handle._pointer, notifyWhen.index,
-        _realmLib.addresses.realm_dart_sync_on_subscription_state_changed_callback, userdata.cast(), _realmLib.addresses.realm_dart_userdata_async_free);
+  Future<SubscriptionSetState> waitForSubscriptionSetStateChange(SubscriptionSet subscriptions, SubscriptionSetState notifyWhen,
+      [CancellationToken? cancellationToken]) {
+    final completer = CancellableCompleter<SubscriptionSetState>(cancellationToken);
+    if (!completer.isCancelled) {
+      final callback = Pointer.fromFunction<Void Function(Handle, Int32)>(_stateChangeCallback);
+      final userdata = _realmLib.realm_dart_userdata_async_new(completer, callback.cast(), scheduler.handle._pointer);
+      _realmLib.realm_sync_on_subscription_set_state_change_async(subscriptions.handle._pointer, notifyWhen.index,
+          _realmLib.addresses.realm_dart_sync_on_subscription_state_changed_callback, userdata.cast(), _realmLib.addresses.realm_dart_userdata_async_free);
+    }
     return completer.future;
   }
 
@@ -666,23 +671,26 @@ class _RealmCore {
 
   Future<RealmHandle> openRealmAsync(RealmAsyncOpenTaskHandle handle, CancellationToken? cancellationToken) {
     final completer = CancellableCompleter<RealmHandle>(cancellationToken);
-    final callback =
-        Pointer.fromFunction<Void Function(Handle, Pointer<realm_thread_safe_reference> realm, Pointer<realm_async_error_t> error)>(_openRealmAsyncCallback);
-    final userData = _realmLib.realm_dart_userdata_async_new(completer, callback.cast(), scheduler.handle._pointer);
-    _realmLib.realm_async_open_task_start(
-      handle._pointer,
-      _realmLib.addresses.realm_dart_async_open_task_callback,
-      userData.cast(),
-      _realmLib.addresses.realm_dart_userdata_async_free,
-    );
-
+    if (!completer.isCancelled) {
+      final callback =
+          Pointer.fromFunction<Void Function(Handle, Pointer<realm_thread_safe_reference> realm, Pointer<realm_async_error_t> error)>(_openRealmAsyncCallback);
+      final userData = _realmLib.realm_dart_userdata_async_new(completer, callback.cast(), scheduler.handle._pointer);
+      _realmLib.realm_async_open_task_start(
+        handle._pointer,
+        _realmLib.addresses.realm_dart_async_open_task_callback,
+        userData.cast(),
+        _realmLib.addresses.realm_dart_userdata_async_free,
+      );
+    }
     return completer.future;
   }
 
   static void _openRealmAsyncCallback(Object userData, Pointer<realm_thread_safe_reference> realmSafePtr, Pointer<realm_async_error_t> error) {
     return using((Arena arena) {
-      final completer = userData as Completer<RealmHandle>;
-
+      final completer = userData as CancellableCompleter<RealmHandle>;
+      if (completer.isCompleted || completer.isCancelled) {
+        return;
+      }
       if (error != nullptr) {
         final err = arena<realm_error>();
         _realmLib.realm_get_async_error(error, err);
@@ -2308,12 +2316,14 @@ class _RealmCore {
     controller.onConnectionStateChange(ConnectionState.values[oldState], ConnectionState.values[newState]);
   }
 
-  Future<void> sessionWaitForUpload(Session session) {
-    final completer = Completer<void>();
-    final callback = Pointer.fromFunction<Void Function(Handle, Pointer<realm_sync_error_code_t>)>(_sessionWaitCompletionCallback);
-    final userdata = _realmLib.realm_dart_userdata_async_new(completer, callback.cast(), scheduler.handle._pointer);
-    _realmLib.realm_sync_session_wait_for_upload_completion(session.handle._pointer, _realmLib.addresses.realm_dart_sync_wait_for_completion_callback,
-        userdata.cast(), _realmLib.addresses.realm_dart_userdata_async_free);
+  Future<void> sessionWaitForUpload(Session session, [CancellationToken? cancellationToken]) {
+    final completer = CancellableCompleter<void>(cancellationToken);
+    if (!completer.isCancelled) {
+      final callback = Pointer.fromFunction<Void Function(Handle, Pointer<realm_sync_error_code_t>)>(_sessionWaitCompletionCallback);
+      final userdata = _realmLib.realm_dart_userdata_async_new(completer, callback.cast(), scheduler.handle._pointer);
+      _realmLib.realm_sync_session_wait_for_upload_completion(session.handle._pointer, _realmLib.addresses.realm_dart_sync_wait_for_completion_callback,
+          userdata.cast(), _realmLib.addresses.realm_dart_userdata_async_free);
+    }
     return completer.future;
   }
 
@@ -2329,8 +2339,8 @@ class _RealmCore {
   }
 
   static void _sessionWaitCompletionCallback(Object userdata, Pointer<realm_sync_error_code_t> errorCode) {
-    final completer = userdata as Completer<void>;
-    if (completer.isCompleted) {
+    final completer = userdata as CancellableCompleter<void>;
+    if (completer.isCompleted || completer.isCancelled) {
       return;
     }
     if (errorCode != nullptr) {

lib/src/realm_class.dart

@@ -34,7 +34,7 @@ import 'session.dart';
 import 'subscription.dart';
 import 'set.dart';
 
-export 'package:cancellation_token/cancellation_token.dart' show CancellationToken, CancelledException;
+export 'package:cancellation_token/cancellation_token.dart' show CancellationToken, TimeoutCancellationToken, CancelledException;
 export 'package:realm_common/realm_common.dart'
     show
         Backlink,
@@ -106,7 +106,7 @@ export 'realm_object.dart'
         RealmObjectChanges,
         UserCallbackException;
 export 'realm_property.dart';
-export 'results.dart' show RealmResultsOfObject, RealmResultsChanges, RealmResults;
+export 'results.dart' show RealmResultsOfObject, RealmResultsChanges, RealmResults, WaitForSyncMode, RealmResultsOfRealmObject;
 export 'session.dart'
     show
         ConnectionStateChange,

lib/src/results.dart

@@ -18,6 +18,8 @@
 import 'dart:async';
 import 'dart:ffi';
 
+import 'package:cancellation_token/cancellation_token.dart';
+
 import 'collections.dart';
 import 'native/realm_core.dart';
 import 'realm_class.dart';
@@ -135,6 +137,99 @@ extension RealmResultsOfObject<T extends RealmObjectBase> on RealmResults<T> {
   }
 }
 
+class _SubscribedRealmResult<T extends RealmObject> extends RealmResults<T> {
+  final String? subscriptionName;
+
+  _SubscribedRealmResult._(RealmResults<RealmObject> results, {this.subscriptionName})
+      : super._(
+          results.handle,
+          results.realm,
+          results.metadata,
+        );
+}
+
+extension RealmResultsOfRealmObject<T extends RealmObject> on RealmResults<T> {
+  /// Adds this [RealmResults] query to the set of active subscriptions.
+  /// The query will be joined via an OR statement with any existing queries for the same type.
+  ///
+  /// If a [name] is given this allows you to later refer to the subscription by name,
+  /// e.g. when calling [MutableSubscriptionSet.removeByName].
+  ///
+  /// If [update] is specified to `true`, then any existing query
+  /// with the same name will be replaced.
+  /// Otherwise a [RealmException] is thrown, in case of duplicates.
+  ///
+  /// [WaitForSyncMode] specifies how to wait or not wait for subscribed objects to be downloaded.
+  /// The default value is [WaitForSyncMode.firstTime].
+  ///
+  /// The [cancellationToken] is optional and can be used to cancel
+  /// the waiting for objects to be downloaded.
+  /// If the operation is cancelled, a [CancelledException] is thrown and the download
+  /// continues in the background.
+  /// In case of using [TimeoutCancellationToken] and the time limit is exceeded,
+  /// a [TimeoutException] is thrown and the download continues in the background.
+  ///
+  /// {@category Sync}
+  Future<RealmResults<T>> subscribe({
+    String? name,
+    WaitForSyncMode waitForSyncMode = WaitForSyncMode.firstTime,
+    CancellationToken? cancellationToken,
+    bool update = false,
+  }) async {
+    Subscription? existingSubscription = name == null ? realm.subscriptions.find(this) : realm.subscriptions.findByName(name);
+    late Subscription updatedSubscription;
+    realm.subscriptions.update((mutableSubscriptions) {
+      updatedSubscription = mutableSubscriptions.add(this, name: name, update: update);
+    });
+    bool shouldWait = waitForSyncMode == WaitForSyncMode.always ||
+        (waitForSyncMode == WaitForSyncMode.firstTime && subscriptionIsChanged(existingSubscription, updatedSubscription));
+
+    return await CancellableFuture.from<RealmResults<T>>(() async {
+      if (cancellationToken != null && cancellationToken.isCancelled) {
+        throw cancellationToken.exception!;
+      }
+      if (shouldWait) {
+        await realm.subscriptions.waitForSynchronization(cancellationToken);
+        await realm.syncSession.waitForDownload(cancellationToken);
+      }
+      return _SubscribedRealmResult._(this, subscriptionName: name);
+    }, cancellationToken);
+  }
+
+  /// Unsubscribe from this query result. It returns immediately
+  /// without waiting for synchronization.
+  ///
+  /// If the subscription is unnamed, the subscription matching
+  /// the query will be removed.
+  /// Return `false` if the [RealmResults] is not created by [subscribe].
+  ///
+  /// {@category Sync}
+  bool unsubscribe() {
+    bool unsubscribed = false;
+    if (realm.config is! FlexibleSyncConfiguration) {
+      throw RealmError('unsubscribe is only allowed on Realms opened with a FlexibleSyncConfiguration');
+    }
+    if (this is _SubscribedRealmResult<T>) {
+      final subscriptionName = (this as _SubscribedRealmResult<T>).subscriptionName;
+      realm.subscriptions.update((mutableSubscriptions) {
+        if (subscriptionName != null) {
+          unsubscribed = mutableSubscriptions.removeByName(subscriptionName);
+        } else {
+          unsubscribed = mutableSubscriptions.removeByQuery(this);
+        }
+      });
+    }
+    return unsubscribed;
+  }
+
+  bool subscriptionIsChanged(Subscription? existingSubscription, Subscription updatedSubscription) {
+    bool changed = existingSubscription == null ||
+        existingSubscription.objectClassName != updatedSubscription.objectClassName ||
+        existingSubscription.queryString != updatedSubscription.queryString;
+    return changed;
+  }
+}
+
 /// @nodoc
 //RealmResults package internal members
 extension RealmResultsInternal on RealmResults {
@@ -231,3 +326,20 @@ class _RealmResultsIterator<T extends Object?> implements Iterator<T> {
     return true;
   }
 }
+
+///
+/// Behavior when waiting for subscribed objects to be synchronized/downloaded.
+///
+enum WaitForSyncMode {
+  /// Waits until the objects have been downloaded from the server
+  /// the first time the subscription is created. If the subscription
+  /// already exists, the [RealmResults] is returned immediately.
+  firstTime,
+
+  /// Always waits until the objects have been downloaded from the server.
+  always,
+
+  /// Never waits for the download to complete, but keeps downloading the
+  /// objects in the background.
+  never,
+}

lib/src/session.dart

@@ -57,9 +57,11 @@ class Session implements Finalizable {
   void resume() => realmCore.sessionResume(this);
 
   /// Waits for the [Session] to finish all pending uploads.
-  Future<void> waitForUpload() => realmCore.sessionWaitForUpload(this);
+  /// An optional [cancellationToken] can be used to cancel the wait operation.
+  Future<void> waitForUpload([CancellationToken? cancellationToken]) => realmCore.sessionWaitForUpload(this, cancellationToken);
 
   /// Waits for the [Session] to finish all pending downloads.
+  /// An optional [cancellationToken] can be used to cancel the wait operation.
   Future<void> waitForDownload([CancellationToken? cancellationToken]) => realmCore.sessionWaitForDownload(this, cancellationToken);
 
   /// Gets a [Stream] of [SyncProgress] that can be used to track upload or download progress.

lib/src/subscription.dart

@@ -147,8 +147,8 @@ abstract class SubscriptionSet with IterableMixin<Subscription> implements Final
     return result == null ? null : Subscription._(result);
   }
 
-  Future<SubscriptionSetState> _waitForStateChange(SubscriptionSetState state) async {
-    final result = await realmCore.waitForSubscriptionSetStateChange(this, state);
+  Future<SubscriptionSetState> _waitForStateChange(SubscriptionSetState state, [CancellationToken? cancellationToken]) async {
+    final result = await realmCore.waitForSubscriptionSetStateChange(this, state, cancellationToken);
     realmCore.refreshSubscriptionSet(this);
     return result;
   }
@@ -159,8 +159,9 @@ abstract class SubscriptionSet with IterableMixin<Subscription> implements Final
   /// the returned [Future] will complete immediately. If the state is
   /// [SubscriptionSetState.error], the returned future will throw an
   /// error.
-  Future<void> waitForSynchronization() async {
-    final result = await _waitForStateChange(SubscriptionSetState.complete);
+  /// An optional [cancellationToken] can be used to cancel the wait operation.
+  Future<void> waitForSynchronization([CancellationToken? cancellationToken]) async {
+    final result = await _waitForStateChange(SubscriptionSetState.complete, cancellationToken);
     if (result == SubscriptionSetState.error) {
       throw error!;
     }
@@ -272,21 +273,22 @@ class MutableSubscriptionSet extends SubscriptionSet {
     return Subscription._(realmCore.insertOrAssignSubscription(this, query, name, update));
   }
 
-  /// Remove the [subscription] from the set, if it exists.
+  /// Removes the [subscription] from the set, if it exists.
   bool remove(Subscription subscription) {
     return realmCore.eraseSubscriptionById(this, subscription);
   }
 
-  /// Remove the [query] from the set, if it exists.
+  /// Removes the [query] from the set, if it exists.
   bool removeByQuery<T extends RealmObject>(RealmResults<T> query) {
     return realmCore.eraseSubscriptionByResults(this, query);
   }
 
-  /// Remove the [query] from the set that matches by [name], if it exists.
+  /// Remove the subscription from the set that matches by [name], if it exists.
   bool removeByName(String name) {
     return realmCore.eraseSubscriptionByName(this, name);
   }
 
+  /// Removes the subscriptions from the set that matches by type, if it exists.
   bool removeByType<T extends RealmObject>() {
     final name = realm.schema.singleWhere((e) => e.type == T).name;
     var result = false;
@@ -301,8 +303,17 @@ class MutableSubscriptionSet extends SubscriptionSet {
   }
 
   /// Clear the subscription set.
-  void clear() {
-    realmCore.clearSubscriptionSet(this);
+  void clear({bool unnamedOnly = false}) {
+    if (unnamedOnly) {
+      for (var i = length - 1; i >= 0; i--) {
+        final subscription = this[i];
+        if (subscription.name == null) {
+          remove(subscription);
+        }
+      }
+    } else {
+      realmCore.clearSubscriptionSet(this);
+    }
   }
 }
 

test/realm_test.dart

@@ -26,7 +26,6 @@ import 'package:test/test.dart' hide test, throws;
 import 'package:timezone/timezone.dart' as tz;
 import 'package:timezone/data/latest.dart' as tz;
 import 'package:path/path.dart' as p;
-import 'package:cancellation_token/cancellation_token.dart';
 import '../lib/realm.dart';
 import 'test.dart';
 import '../lib/src/native/realm_core.dart';
@@ -1017,7 +1016,7 @@ Future<void> main([List<String>? args]) async {
     Future<void>.delayed(Duration(milliseconds: 10), () => transaction.commit());
 
     final transaction1 = await realm.beginWriteAsync();
-    
+
     await transaction1.commitAsync();
 
     expect(transaction.isOpen, false);
@@ -1978,8 +1977,8 @@ Future<void> _addDataToAtlas(Realm realm, String productNamePrefix, {int itemsCo
   await realm.syncSession.waitForDownload();
 }
 
-Future<void> _addSubscriptions(Realm realm, String searchByPreffix) async {
-  final query = realm.query<Product>(r'name BEGINSWITH $0', [searchByPreffix]);
+Future<void> _addSubscriptions(Realm realm, String searchByPrefix) async {
+  final query = realm.query<Product>(r'name BEGINSWITH $0', [searchByPrefix]);
   if (realm.subscriptions.find(query) == null) {
     realm.subscriptions.update((mutableSubscriptions) => mutableSubscriptions.add(query));
   }

test/session_test.dart

@@ -137,6 +137,19 @@ Future<void> main([List<String>? args]) async {
     await realm.syncSession.waitForDownload();
   });
 
+  baasTest('SyncSession.waitForDownload/waitForUpload canceled', (configuration) async {
+    final realm = await getIntegrationRealm();
+    final cancellationDownloadToken = CancellationToken();
+    final waitForDownloadFuture = realm.syncSession.waitForDownload(cancellationDownloadToken);
+    cancellationDownloadToken.cancel();
+    expect(() async => await waitForDownloadFuture, throwsA(isA<CancelledException>()));
+
+    final cancellationUploadToken = CancellationToken();
+    final waitForUploadFuture = realm.syncSession.waitForUpload(cancellationUploadToken);
+    cancellationUploadToken.cancel();
+    expect(() async => await waitForUploadFuture, throwsA(isA<CancelledException>()));
+  });
+
   baasTest('SyncSesison.waitForUpload with changes', (configuration) async {
     final differentiator = ObjectId();