queryListStream<T> method

Stream<List<T>> queryListStream<T>(String sql, { List<Object>? arguments, required String queryableName, required bool isView, required T mapper(Map<String, Object?>), })

Executes a SQLite query that returns a stream of multiple query results.

Implementation

Stream<List<T>> queryListStream<T>(
  final String sql, {
  final List<Object>? arguments,
  required final String queryableName,
  required final bool isView,
  required final T Function(Map<String, Object?>) mapper,
}) {
  // ignore: close_sinks
  final changeListener = ArgumentError.checkNotNull(_changeListener);
  final controller = StreamController<List<T>>.broadcast();

  Future<void> executeQueryAndNotifyController() async {
    final result = await queryList(sql, arguments: arguments, mapper: mapper);
    controller.add(result);
  }

  controller.onListen = () async => executeQueryAndNotifyController();

  // Views listen on all events, Entities only on events that changed the same entity.
  final subscription = changeListener.stream
      .where((updatedTable) =>
          isView || updatedTable.equals(queryableName, ignoreCase: true))
      .listen(
        (_) async => executeQueryAndNotifyController(),
        onDone: () => controller.close(),
      );

  controller.onCancel = () => subscription.cancel();

  return controller.stream;
}