diff --git a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/FlutterFirebaseFirestorePlugin.java b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/FlutterFirebaseFirestorePlugin.java index e162e6530a52..319571602041 100644 --- a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/FlutterFirebaseFirestorePlugin.java +++ b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/FlutterFirebaseFirestorePlugin.java @@ -57,12 +57,17 @@ import io.flutter.plugins.firebase.firestore.utils.PipelineParser; import java.util.ArrayList; import java.util.HashMap; +import java.util.Iterator; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; public class FlutterFirebaseFirestorePlugin @@ -90,6 +95,7 @@ public class FlutterFirebaseFirestorePlugin private final Map eventChannels = new HashMap<>(); private final Map streamHandlers = new HashMap<>(); private final Map transactionHandlers = new HashMap<>(); + private final Map snapshotDeliveryExecutors = new ConcurrentHashMap<>(); // Used in the decoder to know which ServerTimestampBehavior to use public static final Map @@ -116,15 +122,30 @@ protected static void setCachedFirebaseFirestoreInstanceForKey( protected static FirebaseFirestore getFirestoreInstanceByNameAndDatabaseUrl( String appName, String databaseURL) { synchronized (firestoreInstanceCache) { - for (Map.Entry entry : - firestoreInstanceCache.entrySet()) { - if (entry.getValue().getInstance().getApp().getName().equals(appName) + Iterator> iterator = + firestoreInstanceCache.entrySet().iterator(); + FirebaseFirestore match = null; + while (iterator.hasNext()) { + Map.Entry entry = iterator.next(); + // getName() throws after delete(), and the SDK has no public check. + final String cachedAppName; + try { + cachedAppName = entry.getValue().getInstance().getApp().getName(); + } catch (IllegalStateException exception) { + if ("FirebaseApp was deleted".equals(exception.getMessage())) { + iterator.remove(); + continue; + } + throw exception; + } + if (match == null + && cachedAppName.equals(appName) && entry.getValue().getDatabaseURL().equals(databaseURL)) { - return entry.getKey(); + match = entry.getKey(); } } + return match; } - return null; } private static void destroyCachedFirebaseFirestoreInstanceForKey(FirebaseFirestore firestore) { @@ -360,6 +381,15 @@ public static FirebaseFirestore getFirestoreFromPigeon( } } + private ExecutorService snapshotDeliveryExecutor( + GeneratedAndroidFirebaseFirestore.FirestorePigeonFirebaseApp app) { + String key = app.getAppName() + "|" + app.getDatabaseURL(); + return snapshotDeliveryExecutors.computeIfAbsent( + key, + ignored -> + new ThreadPoolExecutor(0, 1, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>())); + } + @Override public void loadBundle( @NonNull GeneratedAndroidFirebaseFirestore.FirestorePigeonFirebaseApp app, @@ -563,7 +593,8 @@ public void snapshotsInSyncSetup( result.success( registerEventChannel( - METHOD_CHANNEL_NAME + "/snapshotsInSync", new SnapshotsInSyncStreamHandler(firestore))); + METHOD_CHANNEL_NAME + "/snapshotsInSync", + new SnapshotsInSyncStreamHandler(firestore, snapshotDeliveryExecutor(app)))); } @Override @@ -1019,7 +1050,7 @@ public void querySnapshot( PigeonParser.parsePigeonServerTimestampBehavior( options.getServerTimestampBehavior()), PigeonParser.parseListenSource(source), - cachedThreadPool))); + snapshotDeliveryExecutor(app)))); } @Override diff --git a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/SnapshotsInSyncStreamHandler.java b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/SnapshotsInSyncStreamHandler.java index 8d58f40ef8cb..4303daae7264 100644 --- a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/SnapshotsInSyncStreamHandler.java +++ b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/SnapshotsInSyncStreamHandler.java @@ -6,25 +6,31 @@ package io.flutter.plugins.firebase.firestore.streamhandler; +import android.os.Handler; +import android.os.Looper; import com.google.firebase.firestore.FirebaseFirestore; import com.google.firebase.firestore.ListenerRegistration; import io.flutter.plugin.common.EventChannel.EventSink; import io.flutter.plugin.common.EventChannel.StreamHandler; +import java.util.concurrent.Executor; public class SnapshotsInSyncStreamHandler implements StreamHandler { ListenerRegistration listenerRegistration; FirebaseFirestore firestore; + private final Executor snapshotExecutor; + private final Handler mainHandler = new Handler(Looper.getMainLooper()); - public SnapshotsInSyncStreamHandler(FirebaseFirestore firestore) { + public SnapshotsInSyncStreamHandler(FirebaseFirestore firestore, Executor snapshotExecutor) { this.firestore = firestore; + this.snapshotExecutor = snapshotExecutor; } @Override public void onListen(Object arguments, EventSink events) { - Runnable snapshotsInSyncRunnable = () -> events.success(null); - - listenerRegistration = firestore.addSnapshotsInSyncListener(snapshotsInSyncRunnable); + listenerRegistration = + firestore.addSnapshotsInSyncListener( + snapshotExecutor, () -> mainHandler.post(() -> events.success(null))); } @Override diff --git a/packages/cloud_firestore/cloud_firestore/example/integration_test/instance_e2e.dart b/packages/cloud_firestore/cloud_firestore/example/integration_test/instance_e2e.dart index e589c02d4ae8..ee6cade3019c 100644 --- a/packages/cloud_firestore/cloud_firestore/example/integration_test/instance_e2e.dart +++ b/packages/cloud_firestore/cloud_firestore/example/integration_test/instance_e2e.dart @@ -62,6 +62,109 @@ void runInstanceTests() { await snapshots.cancel(); }, skip: kIsWeb); + test( + 'snapshotsInSync() follows a remote query snapshot', + () async { + final writerApp = await Firebase.initializeApp( + name: 'insync-writer-${DateTime.now().microsecondsSinceEpoch}', + options: Firebase.app().options, + ); + final writer = FirebaseFirestore.instanceFor(app: writerApp); + writer.settings = const Settings(persistenceEnabled: false); + writer.useFirestoreEmulator('localhost', 8080); + + final collection = firestore + .collection('flutter-tests') + .doc('insync-${DateTime.now().microsecondsSinceEpoch}') + .collection('items'); + const docId = 'shared'; + final events = []; + + final querySub = collection.snapshots().listen((snapshot) { + String? value; + for (final document in snapshot.docs) { + if (document.id == docId) { + value = '${document.data()['n']}'; + } + } + events.add('query:$value:${snapshot.metadata.hasPendingWrites}'); + }); + final syncSub = firestore.snapshotsInSync().listen((_) { + events.add('sync'); + }); + + Future waitUntil(bool Function() ready) async { + final deadline = DateTime.now().add(const Duration(seconds: 5)); + while (!ready()) { + expect( + DateTime.now().isBefore(deadline), + isTrue, + reason: '$events', + ); + await Future.delayed(const Duration(milliseconds: 20)); + } + } + + Future settle() async { + final deadline = DateTime.now().add(const Duration(seconds: 2)); + var count = events.length; + while (DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 80)); + if (events.length == count) { + return; + } + count = events.length; + } + } + + await waitUntil( + () => + events.contains('sync') && + events.any((event) => event.startsWith('query:')), + ); + await settle(); + events.clear(); + + for (var i = 1; i <= 24; i++) { + final marker = events.length; + await writer.collection(collection.path).doc(docId).set({'n': i}); + await waitUntil( + () => events.sublist(marker).contains('query:$i:false'), + ); + await waitUntil(() => events.sublist(marker).contains('sync')); + final slice = events.sublist(marker); + final queryAt = slice.indexOf('query:$i:false'); + final syncAt = slice.indexOf('sync'); + expect(queryAt, lessThan(syncAt), reason: 'write $i: $slice'); + await settle(); + } + + await querySub.cancel(); + await syncSub.cancel(); + await writerApp.delete(); + + // This database is not cached, so the lookup has to walk every + // entry, including the deleted writer's. + final uncached = FirebaseFirestore.instanceFor( + app: Firebase.app(), + databaseId: 'insync-after-delete', + ); + uncached.settings = const Settings(persistenceEnabled: false); + uncached.useFirestoreEmulator('localhost', 8080); + await uncached.doc('flutter-tests/insync-after-delete').set({ + 'ok': true, + }); + await firestore.doc('flutter-tests/insync-after-delete').set({ + 'ok': true, + }); + }, + timeout: const Timeout(Duration(minutes: 2)), + skip: + defaultTargetPlatform != TargetPlatform.iOS && + defaultTargetPlatform != TargetPlatform.macOS && + defaultTargetPlatform != TargetPlatform.android, + ); + test('enableNetwork()', () async { // Write some data while online await firestore.enableNetwork(); diff --git a/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/FLTFirebaseFirestorePlugin.swift b/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/FLTFirebaseFirestorePlugin.swift index e77bc9bf1d8a..68e9e470e0ab 100644 --- a/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/FLTFirebaseFirestorePlugin.swift +++ b/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/FLTFirebaseFirestorePlugin.swift @@ -35,6 +35,8 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu private var transactionHandlers: [String: TransactionStreamHandler] = [:] private let transactionLock = NSLock() private let listenersLock = NSLock() + private var snapshotDeliveryQueues: [String: DispatchQueue] = [:] + private let snapshotDeliveryQueuesLock = NSLock() static let serverTimestampMap = NSCache() @@ -153,6 +155,20 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu return identifier } + private func snapshotDeliveryQueue(appName: String, databaseURL: String) -> DispatchQueue { + let key = "\(appName)|\(databaseURL)" + snapshotDeliveryQueuesLock.lock() + defer { snapshotDeliveryQueuesLock.unlock() } + if let queue = snapshotDeliveryQueues[key] { + return queue + } + let queue = DispatchQueue( + label: "io.flutter.plugins.firebase.firestore.snapshot_delivery.\(key)" + ) + snapshotDeliveryQueues[key] = queue + return queue + } + private func firestore(from pigeonApp: FirestorePigeonFirebaseApp) -> Firestore { objc_sync_enter(self) defer { objc_sync_exit(self) } @@ -353,7 +369,10 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu let firestore = firestore(from: app) let identifier = registerEventChannel( prefix: kFLTFirebaseFirestoreSnapshotsInSyncEventChannelName, - streamHandler: SnapshotsInSyncStreamHandler(firestore: firestore) + streamHandler: SnapshotsInSyncStreamHandler( + firestore: firestore, + snapshotQueue: snapshotDeliveryQueue(appName: app.appName, databaseURL: app.databaseURL) + ) ) completion(.success(identifier)) } @@ -732,7 +751,8 @@ public class FLTFirebaseFirestorePlugin: NSObject, FlutterPlugin, FLTFirebasePlu serverTimestampBehavior: PigeonParser.parseServerTimestampBehavior( options.serverTimestampBehavior ), - source: PigeonParser.parseListenSource(source) + source: PigeonParser.parseListenSource(source), + snapshotQueue: snapshotDeliveryQueue(appName: app.appName, databaseURL: app.databaseURL) ) ) completion(.success(identifier)) diff --git a/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/QuerySnapshotStreamHandler.swift b/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/QuerySnapshotStreamHandler.swift index 5a38338e808a..adca8c8ef9cd 100644 --- a/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/QuerySnapshotStreamHandler.swift +++ b/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/QuerySnapshotStreamHandler.swift @@ -24,22 +24,22 @@ final class QuerySnapshotStreamHandler: NSObject, FlutterStreamHandler { private let serverTimestampBehavior: FirebaseFirestore.ServerTimestampBehavior private let source: FirebaseFirestore.ListenSource private var listenerRegistration: ListenerRegistration? - private let snapshotQueue = DispatchQueue( - label: "io.flutter.plugins.firebase.firestore.query_snapshot" - ) + private let snapshotQueue: DispatchQueue init( firestore: Firestore, query: Query?, includeMetadataChanges: Bool, serverTimestampBehavior: FirebaseFirestore.ServerTimestampBehavior, - source: FirebaseFirestore.ListenSource + source: FirebaseFirestore.ListenSource, + snapshotQueue: DispatchQueue ) { self.firestore = firestore self.query = query self.includeMetadataChanges = includeMetadataChanges self.serverTimestampBehavior = serverTimestampBehavior self.source = source + self.snapshotQueue = snapshotQueue } func onListen(withArguments arguments: Any?, eventSink events: @escaping FlutterEventSink) diff --git a/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/SnapshotsInSyncStreamHandler.swift b/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/SnapshotsInSyncStreamHandler.swift index 88328f057450..b7fdc80906a8 100644 --- a/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/SnapshotsInSyncStreamHandler.swift +++ b/packages/cloud_firestore/cloud_firestore/ios/cloud_firestore/Sources/cloud_firestore/SnapshotsInSyncStreamHandler.swift @@ -13,18 +13,22 @@ import Foundation final class SnapshotsInSyncStreamHandler: NSObject, FlutterStreamHandler { private let firestore: Firestore + private let snapshotQueue: DispatchQueue private var listenerRegistration: ListenerRegistration? - init(firestore: Firestore) { + init(firestore: Firestore, snapshotQueue: DispatchQueue) { self.firestore = firestore + self.snapshotQueue = snapshotQueue } func onListen(withArguments arguments: Any?, eventSink events: @escaping FlutterEventSink) -> FlutterError? { - listenerRegistration = firestore.addSnapshotsInSyncListener { - DispatchQueue.main.async { - events(nil) + listenerRegistration = firestore.addSnapshotsInSyncListener { [snapshotQueue] in + snapshotQueue.async { + DispatchQueue.main.async { + events(nil) + } } } return nil