<commit_before>import 'dart:async';

import 'package:collection/collection.dart';
import 'package:meta/meta.dart';
import 'package:moor/moor.dart';

const _listEquality = ListEquality<dynamic>();

/// Representation of a select statement that knows from which tables the
/// statement is reading its data and how to execute the query.
class QueryStreamFetcher<T> {
  /// The set of tables this query reads from. If any of these tables changes,
  /// the stream must fetch its again.
  final Set<TableInfo> readsFrom;

  /// Key that can be used to check whether two fetchers will yield the same
  /// result when operating on the same data.
  final StreamKey key;

  /// Function that asynchronously fetches the latest set of data.
  final Future<T> Function() fetchData;

  QueryStreamFetcher(
      {@required this.readsFrom, this.key, @required this.fetchData});
}

/// Key that uniquely identifies a select statement. If two keys created from
/// two select statements are equal, the statements are equal as well.
///
/// As two equal statements always yield the same result when operating on the
/// same data, this can make streams more efficient as we can return the same
/// stream for two equivalent queries.
class StreamKey {
  final String sql;
  final List<dynamic> variables;

  /// Used to differentiate between custom streams, which return a [QueryRow],
  /// and regular streams, which return an instance of a generated data class.
  final Type returnType;

  StreamKey(this.sql, this.variables, this.returnType);

  @override
  int get hashCode {
    return (sql.hashCode * 31 + _listEquality.hash(variables)) * 31 +
        returnType.hashCode;
  }

  @override
  bool operator ==(other) {
    return identical(this, other) ||
        (other is StreamKey &&
            other.sql == sql &&
            _listEquality.equals(other.variables, variables) &&
            other.returnType == returnType);
  }
}

/// Keeps track of active streams created from [SimpleSelectStatement]s and updates
/// them when needed.
class StreamQueryStore {
  final List<QueryStream> _activeStreamsWithoutKey = [];
  final Map<StreamKey, QueryStream> _activeKeyStreams = {};

  StreamQueryStore();

  Iterable<QueryStream> get _activeStreams {
    return _activeKeyStreams.values.followedBy(_activeStreamsWithoutKey);
  }

  /// Creates a new stream from the select statement.
  Stream<T> registerStream<T>(QueryStreamFetcher<T> fetcher) {
    final key = fetcher.key;

    if (key == null) {
      final stream = QueryStream(fetcher, this);
      _activeStreamsWithoutKey.add(stream);
      return stream.stream;
    } else {
      final stream = _activeKeyStreams.putIfAbsent(key, () {
        return QueryStream<T>(fetcher, this);
      });

      return (stream as QueryStream<T>).stream;
    }
  }

  /// Handles updates on a given table by re-executing all queries that read
  /// from that table.
  Future<void> handleTableUpdates(Set<TableInfo> tables) async {
    final affectedStreams = _activeStreams
        .where((stream) => stream._fetcher.readsFrom.any(tables.contains));

    for (var stream in affectedStreams) {
      await stream.fetchAndEmitData();
    }
  }

  void markAsClosed(QueryStream stream) {
    final key = stream._fetcher.key;
    if (key == null) {
      _activeStreamsWithoutKey.remove(stream);
    } else {
      _activeKeyStreams.remove(key);
    }
  }
}

class QueryStream<T> {
  final QueryStreamFetcher<T> _fetcher;
  final StreamQueryStore _store;

  StreamController<T> _controller;
  // caching the stream so that the stream getter always returns the same stream
  Stream<T> _stream;

  Stream<T> get stream {
    _controller ??= StreamController.broadcast(
      onListen: _onListen,
      onCancel: _onCancel,
    );

    return _stream ??= _controller.stream;
  }

  QueryStream(this._fetcher, this._store);

  void _onListen() {
    // first listener added, fetch query
    fetchAndEmitData();
  }

  void _onCancel() {
    // last listener gone, dispose
    _controller.close();
    // todo this removes the stream from the list so that it can be garbage
    // collected. When a stream is never listened to, we have a memory leak as
    // this will never be called. Maybe an Expando (which uses weak references)
    // can save us here?
    _store.markAsClosed(this);
  }

  Future<void> fetchAndEmitData() async {
    // Fetch data if it's needed, publish that data if it's possible.
    if (!_controller.hasListener) return;

    final data = await _fetcher.fetchData();

    if (!_controller.isClosed) {
      _controller.add(data);
    }
  }
}
<commit_msg>Fix concurrent modification in stream query store<commit_after>import 'dart:async';

import 'package:collection/collection.dart';
import 'package:meta/meta.dart';
import 'package:moor/moor.dart';

const _listEquality = ListEquality<dynamic>();

/// Representation of a select statement that knows from which tables the
/// statement is reading its data and how to execute the query.
class QueryStreamFetcher<T> {
  /// The set of tables this query reads from. If any of these tables changes,
  /// the stream must fetch its again.
  final Set<TableInfo> readsFrom;

  /// Key that can be used to check whether two fetchers will yield the same
  /// result when operating on the same data.
  final StreamKey key;

  /// Function that asynchronously fetches the latest set of data.
  final Future<T> Function() fetchData;

  QueryStreamFetcher(
      {@required this.readsFrom, this.key, @required this.fetchData});
}

/// Key that uniquely identifies a select statement. If two keys created from
/// two select statements are equal, the statements are equal as well.
///
/// As two equal statements always yield the same result when operating on the
/// same data, this can make streams more efficient as we can return the same
/// stream for two equivalent queries.
class StreamKey {
  final String sql;
  final List<dynamic> variables;

  /// Used to differentiate between custom streams, which return a [QueryRow],
  /// and regular streams, which return an instance of a generated data class.
  final Type returnType;

  StreamKey(this.sql, this.variables, this.returnType);

  @override
  int get hashCode {
    return (sql.hashCode * 31 + _listEquality.hash(variables)) * 31 +
        returnType.hashCode;
  }

  @override
  bool operator ==(other) {
    return identical(this, other) ||
        (other is StreamKey &&
            other.sql == sql &&
            _listEquality.equals(other.variables, variables) &&
            other.returnType == returnType);
  }
}

/// Keeps track of active streams created from [SimpleSelectStatement]s and updates
/// them when needed.
class StreamQueryStore {
  final List<QueryStream> _activeStreamsWithoutKey = [];
  final Map<StreamKey, QueryStream> _activeKeyStreams = {};

  StreamQueryStore();

  Iterable<QueryStream> get _activeStreams {
    return _activeKeyStreams.values.followedBy(_activeStreamsWithoutKey);
  }

  /// Creates a new stream from the select statement.
  Stream<T> registerStream<T>(QueryStreamFetcher<T> fetcher) {
    final key = fetcher.key;

    if (key == null) {
      final stream = QueryStream(fetcher, this);
      _activeStreamsWithoutKey.add(stream);
      return stream.stream;
    } else {
      final stream = _activeKeyStreams.putIfAbsent(key, () {
        return QueryStream<T>(fetcher, this);
      });

      return (stream as QueryStream<T>).stream;
    }
  }

  /// Handles updates on a given table by re-executing all queries that read
  /// from that table.
  Future<void> handleTableUpdates(Set<TableInfo> tables) async {
    final activeStreams = List<QueryStream>.from(_activeStreams);
    final affectedStreams = activeStreams
        .where((stream) => stream._fetcher.readsFrom.any(tables.contains));

    for (var stream in affectedStreams) {
      await stream.fetchAndEmitData();
    }
  }

  void markAsClosed(QueryStream stream) {
    final key = stream._fetcher.key;
    if (key == null) {
      _activeStreamsWithoutKey.remove(stream);
    } else {
      _activeKeyStreams.remove(key);
    }
  }
}

class QueryStream<T> {
  final QueryStreamFetcher<T> _fetcher;
  final StreamQueryStore _store;

  StreamController<T> _controller;
  // caching the stream so that the stream getter always returns the same stream
  Stream<T> _stream;

  Stream<T> get stream {
    _controller ??= StreamController.broadcast(
      onListen: _onListen,
      onCancel: _onCancel,
    );

    return _stream ??= _controller.stream;
  }

  QueryStream(this._fetcher, this._store);

  void _onListen() {
    // first listener added, fetch query
    fetchAndEmitData();
  }

  void _onCancel() {
    // last listener gone, dispose
    _controller.close();
    // todo this removes the stream from the list so that it can be garbage
    // collected. When a stream is never listened to, we have a memory leak as
    // this will never be called. Maybe an Expando (which uses weak references)
    // can save us here?
    _store.markAsClosed(this);
  }

  Future<void> fetchAndEmitData() async {
    // Fetch data if it's needed, publish that data if it's possible.
    if (!_controller.hasListener) return;

    final data = await _fetcher.fetchData();

    if (!_controller.isClosed) {
      _controller.add(data);
    }
  }
}

{'fraction_non_alphanumeric': 0.09736983360171766, 'fraction_numerical': 0.0008588298443370907, 'mean_word_length': 3.277318640955005, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6640282', 'n_tokens_mistral': 2693, 'n_tokens_neox': 2623, 'n_words': 903}