watch method

@override Stream watch(String sql, { List<Object?> params = const [], FromMap? fromMap, bool singleResult = false, required List<String> tables, String? dbName, })

Returns a Stream that emits results of sql whenever tables change.

The stream automatically re-queries the database after any execute call that touches one of the watched tables.

Implementation

@override
Stream watch(
  String sql, {
  List<Object?> params = const [],
  FromMap? fromMap,
  bool singleResult = false,
  required List<String> tables,
  String? dbName,
}) {
  StreamSubscription? sub;
  final sc = StreamController(
    onCancel: () => sub?.cancel(),
  );

  void startWatch() {
    if (sc.isClosed) return;
    sub?.cancel();

    try {
      final responseStream = client.watch(WatchRequest(
        sql: sql,
        params: convertParamsToParam(params),
        dbName: dbName ?? defaultDBName,
        tables: tables,
        singleResult: singleResult,
      ));

      sub = responseStream.listen(
        (response) {
          if (singleResult) {
            if (response.rows.isNotEmpty) {
              // Single row with multiple columns → Map result
              final map = mapsFromRows(response.rows.toList()).first;
              if (fromMap != null) {
                sc.add(fromMap(map));
              } else {
                sc.add(map);
              }
            } else {
              // Scalar result
              final val = dartFromValue(response.result);
              if (fromMap != null && val is Map<String, dynamic>) {
                sc.add(fromMap(val));
              } else {
                sc.add(val);
              }
            }
          } else {
            final rows = mapsFromRows(response.rows.toList());
            if (fromMap != null) {
              sc.add(rows.map((e) => fromMap(e)).toList());
            } else {
              sc.add(rows);
            }
          }
        },
        onDone: () {
          if (!sc.isClosed) {
            Future.delayed(const Duration(seconds: 1), startWatch);
          }
        },
        onError: (e) {
          if (!sc.isClosed) {
            Future.delayed(const Duration(seconds: 2), startWatch);
          }
        },
        cancelOnError: false,
      );
    } catch (e) {
      if (!sc.isClosed) {
        Future.delayed(const Duration(seconds: 2), startWatch);
      }
    }
  }

  startWatch();
  return sc.stream;
}