onQuery function

Implementation

Stream<List<QueryDocument>> onQuery(
  String collectionPath, {
  List<Where> where = const [],
  List<OrderBy> orderBy = const [],
  int? limit,
  int? limitToLast,
  List<Object?>? startAt,
  List<Object?>? startAfter,
  List<Object?>? endAt,
  List<Object?>? endBefore,

  /// Searches every collection with this id, at any depth, rather than one
  /// collection at a path.
  bool collectionGroup = false,
}) {
  final encoded = _encodeSpec(
    _querySpec(
      where: where,
      orderBy: orderBy,
      limit: limit,
      limitToLast: limitToLast,
      startAt: startAt,
      startAfter: startAfter,
      endAt: endAt,
      endBefore: endBefore,
      collectionGroup: collectionGroup,
    ),
  );

  late StreamController<List<QueryDocument>> controller;
  late RawReceivePort receive;
  var listenerId = 0;

  void stop() {
    if (listenerId > 0) fdbFsUnlisten(listenerId);
    receive.close();
  }

  controller = StreamController<List<QueryDocument>>(
    onCancel: stop,
    onListen: () {
      receive = RawReceivePort();
      receive.handler = (Object? message) {
        final bytes = message! as Uint8List;
        final seq = ByteData.sublistView(bytes).getInt64(8, Endian.host);
        if (seq < 0) {
          // The payload carries the reason rather than just the fact: a
          // listener that stops silently is nearly always a rules problem.
          final reason = decodeSnapshotValue(bytes);
          controller.addError(
            StateError(
              'firestore query listener canceled: ${reason ?? "no reason"}',
            ),
          );
          return;
        }
        controller.add(_decodeQueryResult(bytes));
      };

      final p = collectionPath.toNativeUtf8();
      final buf = calloc<Uint8>(encoded.isEmpty ? 1 : encoded.length);
      if (encoded.isNotEmpty) {
        buf.asTypedList(encoded.length).setAll(0, encoded);
      }
      listenerId = fdbFsQueryListen(
        p.cast(),
        buf,
        encoded.length,
        receive.sendPort.nativePort,
      );
      calloc
        ..free(p)
        ..free(buf);
      if (listenerId < 0) {
        controller.addError(
          listenerId == -3 || listenerId == -4
              ? ArgumentError('this query cannot be watched as expressed')
              : StateError('watch $collectionPath failed ($listenerId)'),
        );
        stop();
      }
    },
  );
  return controller.stream;
}