Sync - make sure Observers (query stream) and SyncClient can't be used at the same time

Author: Ivan Dlugos

example/flutter/objectbox_demo_sync/lib/main.dart

@@ -77,7 +77,11 @@ class ViewModel {
 
   void removeNote(Note note) => _box.remove(note.id);
 
-  Stream<List<Note>> get queryStream => _query.findStream();
+  // Note: using query.findStream() and sync.client() in the same app is
+  // currently not supported so this app is currently not working and only
+  // servers as an example on how and when to start a sync client.
+  // Stream<List<Note>> get queryStream => _query.findStream();
+  Stream<List<Note>> get queryStream => Stream<List<Note>>.empty();
 
   List<Note> get allNotes => _query.find();
 

lib/src/observable.dart

@@ -35,6 +35,8 @@ class _Observable {
   }
 
   static void subscribe(Store store) {
+    syncOrObserversExclusive.mark(store);
+
     final callback = Pointer.fromFunction<obx_observer_t>(_anyCallback);
     final storePtr = store.ptr;
     _anyObserver[storePtr.address] =
@@ -53,18 +55,19 @@ class _Observable {
     StoreCloseObserver.removeListener(store, _anyObserver[storeAddress]);
     bindings.obx_observer_close(_anyObserver[storeAddress]);
     _anyObserver.remove(storeAddress);
+    syncOrObserversExclusive.unmark(store);
   }
+
+  static bool isSubscribed(Store store) =>
+      _Observable._anyObserver.containsKey(store.ptr.address);
 }
 
 extension Streamable<T> on Query<T> {
   void _setup() {
-    final storePtr = store.ptr;
-
-    if (!_Observable._anyObserver.containsKey(storePtr)) {
+    if (!_Observable.isSubscribed(store)) {
       _Observable.subscribe(store);
     }
-
-    final storeAddress = storePtr.address;
+    final storeAddress = store.ptr.address;
 
     _Observable._any[storeAddress] ??= <int, Any>{};
     _Observable._any[storeAddress][entityId] ??= (u, _, __) {

lib/src/sync.dart

@@ -101,6 +101,7 @@ class SyncClient {
     _cSync = nullptr;
     SyncClientsStorage.remove(_store);
     StoreCloseObserver.removeListener(_store, this);
+    syncOrObserversExclusive.unmark(_store);
     checkObx(err);
   }
 
@@ -235,6 +236,7 @@ class Sync {
     if (SyncClientsStorage.containsKey(store)) {
       throw Exception('Only one sync client can be active for a store');
     }
+    syncOrObserversExclusive.mark(store);
     final client = SyncClient(store, serverUri, creds);
     SyncClientsStorage[store] = client;
     StoreCloseObserver.addListener(store, client, client.close);

lib/src/util.dart

@@ -42,3 +42,23 @@ class StoreCloseObserver {
 
 /// Global internal storage of sync clients - one client per store.
 final Map<Store, SyncClient> SyncClientsStorage = {};
+
+// Currently, either SyncClient or Observers can be used at the same time.
+// TODO: lift this condition after #142 is fixed.
+class SyncOrObserversExclusive {
+  final _map = <Store, bool>{};
+
+  void mark(Store store) {
+    if (_map.containsKey(store)) {
+      throw Exception(
+          'Using observers/query streams in combination with SyncClient is currently not supported');
+    }
+    _map[store] = true;
+  }
+
+  void unmark(Store store) {
+    _map.remove(store);
+  }
+}
+
+final syncOrObserversExclusive = SyncOrObserversExclusive();

test/sync_test.dart

@@ -3,6 +3,7 @@ import 'dart:typed_data';
 
 import 'package:test/test.dart';
 import 'package:objectbox/objectbox.dart';
+import 'package:objectbox/observable.dart';
 import 'package:objectbox/src/bindings/constants.dart';
 
 import 'entity.dart';
@@ -56,6 +57,24 @@ void main() {
   if (Sync.isAvailable()) {
     // TESTS to run when SYNC is available
 
+    group('Circumvent issue #142 - async callbacks error', () {
+      final error = throwsA(predicate((Exception e) => e.toString().contains(
+          'Using observers/query streams in combination with SyncClient is currently not supported')));
+
+      test('Must not start an Observer when SyncClient is active', () {
+        createClient(store);
+        expect(() => env.box.query().build().findStream(), error);
+      });
+
+      test('Must not start SyncClient when an Observer is active', () {
+        final error = throwsA(predicate((Exception e) => e.toString().contains(
+            'Using observers/query streams in combination with SyncClient is currently not supported')));
+
+        SyncClient c = createClient(store);
+        expect(() => env.box.query().build().findStream(), error);
+      });
+    });
+    
     test('SyncClient lifecycle', () {
       expect(store.syncClient(), isNull);