watch method

@override QueryStream<List<Row>> watch()

Creates an auto-updating stream of the result that emits new items whenever any table used in this statement changes.

Implementation

@override
QueryStream<List<Row>> watch() {
  final stmt = _database.dialect.compile(this);
  final converter = createMapper(_database.dialect, structure);

  final streams = _database.currentStreamQueryStore();
  final fetcher = QueryStreamFetcher(
    readsFrom: TableUpdateQuery.allOf([
      for (final watched in stmt.watchedTables)
        TableUpdateQuery.onTableName(watched),
    ]),
    key: StreamKey(stmt.sql, stmt.variables),
    fetchData: () async {
      final currentSession = await _database.currentSession();
      return currentSession.execute(stmt);
    },
    prepare: () => _database.currentSession(),
  );
  final raw = streams.registerStream<QueryResult>(fetcher);

  return QueryStream(
    fetcher: fetcher,
    stream: raw.map((rows) => rows.resultSet!.map(converter).toList()),
  );
}