query method

@override Stream<Row> query(String sql, [ dynamic values ])

Queue a SQL query to be run, returning a Stream of rows.

Implementation

@override
Stream<Row> query(String sql, [values]) {
  if (_closed)
    throw StateError("Closed: ${_getErrorMessage(sql, values)}");

  _onQuery?.call(sql, values);

  final controller = StreamController<Row>(),
    tmPreSlow = _startSql();
  conn.query(sql, values)
    .listen((data) => controller.add(data),
      onError: (Object ex, StackTrace st) {
        controller.addError(ex, st);
        tmPreSlow?.cancel();

        if (_shallLogError(this, sql, ex))
          _logger.severe("Failed to query: ${_getErrorMessage(sql, values)}", ex, st);
      },
      onDone: () {
        controller.close();

        _checkSlowSql(sql, values);
        tmPreSlow?.cancel();
      },
      cancelOnError: true);
  return controller.stream;
}