onDocument function

Implementation

Stream<Map<String, Object?>?> onDocument(String path) {
  late StreamController<Map<String, Object?>?> controller;
  late RawReceivePort receive;
  var listenerId = 0;

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

  controller = StreamController<Map<String, Object?>?>(
    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.
          final reason = decodeSnapshotValue(bytes);
          controller.addError(
            StateError('firestore listener canceled: ${reason ?? "no reason"}'),
          );
          return;
        }
        controller.add(decodeDocument(bytes));
      };
      final p = path.toNativeUtf8();
      listenerId = fdbFsListen(p.cast(), receive.sendPort.nativePort);
      calloc.free(p);
      if (listenerId < 0) {
        controller.addError(StateError('listen $path failed ($listenerId)'));
        stop();
      }
    },
  );
  return controller.stream;
}