onValue function

Implementation

Stream<DbSnapshot> onValue(String path) {
  final port = ReceivePort();
  final p = path.toNativeUtf8();
  late final int handle;
  late final StreamController<DbSnapshot> controller;

  void stop() {
    fdbDbUnlisten(handle);
    port.close();
    calloc.free(p);
  }

  controller = StreamController<DbSnapshot>(onCancel: stop);

  handle = fdbDbListen(p.cast(), port.sendPort.nativePort);
  if (handle < 0) {
    calloc.free(p);
    port.close();
    return Stream<DbSnapshot>.error(
      StateError('listen failed: database not initialized'),
    );
  }

  port.listen((message) {
    final bytes = message as Uint8List;
    final view = ByteData.sublistView(bytes);
    if (view.getUint32(0, Endian.host) != fdbSnapshotMagic) {
      controller.addError(const FormatException('bad snapshot magic'));
      return;
    }
    // Offsets into FdbSnapshotHeader: magic 0, version 4, seq 8, posted_ns 16.
    final seq = view.getInt64(8, Endian.host);
    final postedNs = view.getInt64(16, Endian.host);
    if (seq < 0) {
      // The payload carries the SDK's error code and message.
      final reason = decodeSnapshotValue(bytes);
      controller.addError(
        StateError('database listener canceled: ${reason ?? "no reason"}'),
      );
      return;
    }
    controller.add(
      DbSnapshot(
        seq: seq,
        value: decodeSnapshotValue(bytes),
        postedNs: postedNs,
        order: decodeSnapshotOrder(bytes),
      ),
    );
  });

  return controller.stream;
}