onSnapshots method

Stream<List<RecordSnapshot<K, V>>> onSnapshots(Database database)

Find multiple records and listen for changes.

Returns a single subscriber stream that must be cancelled.

Implementation

Stream<List<RecordSnapshot<K, V>>> onSnapshots(Database database) {
  var db = getDatabase(database);
  // Create the query but don't add it until first result is set
  late QueryListenerController<K, V> ctlr;
  ctlr = db.listener.addQuery(
    this,
    onListen: () async {
      // Add the existing snapshot

      // Read right away to get the content at call time

      // Just filter
      try {
        // Make sure the first read matches the existing content.
        await ctlr.lock.synchronized(() async {
          // Find all matching, ignoring offset/limit but order them
          var allMatching = await sembastQueryRef.store.findImmutableRecords(
            database,
            finder:
                sembastQueryRef.finder?.cloneWithoutLimits()
                    as SembastFinder?,
          );
          // ignore: unawaited_futures

          // Get the result at query time first
          if (debugListener) {
            // ignore: avoid_print
            print('matching $ctlr: ${allMatching.length} on $this');
          }

          await ctlr.add(allMatching, db.cooperator);
        });
      } catch (error, stackTrace) {
        ctlr.addError(error, stackTrace);
      }
    },
  );
  return ctlr.stream;
}