<commit_before>part of streamy.runtime;

/// An [EventTransformer] that de-duplicates entities. This will cause
/// metadata about the entity (Entity.streamy) to be inaccurate, but will
/// prevent multiple values from being published on [Stream]s when core [Entity]
/// data has not changed.
class EntityDedupTransformer extends EventTransformer {
  var _last = null;

  EntityDedupTransformer() : super();

  void handleData(Response<T> response, EventSink<T> sink, Trace trace) {
    if (!Entity.deepEquals(response.entity, _last)) {
      sink.add(response);
    }
    _last = response.entity;
  }
}

/// An [EventTransformer] that closes the stream after the RPC reply is
/// received.
class OneShotRequestTransformer extends EventTransformer {

  const OneShotRequestTransformer() : super();

  void handleData(Response response, EventSink<Response> sink, Trace trace) {
    sink.add(response);
    if (response.source == Source.RPC) {
      sink.close();
    }
  }
}

/// An [EventTransformer] that clones frozen entities, to make them mutable.
class MutableTransformer extends EventTransformer {

  const MutableTransformer() : super();

  void handleData(Response response, EventSink<Response> sink, Trace trace) {
    if (response.entity.isFrozen) {
      sink.add(new Response(response.entity.clone(), response.source, response.ts));
    } else {
      sink.add(response);
    }
  }
}

/// An [EventTransformer] that traces user callback timings. Should be the last
/// transformer before user code.
class UserCallbackTracingTransformer extends EventTransformer {

  var _openCallbacks = 0;
  var _closed = false;
  var _sentDone = false;

  UserCallbackTracingTransformer() : super();

  void handleData(Response response, EventSink<Response> sink, Trace trace) {
    trace.record(new UserCallbackQueuedEvent());
    _openCallbacks++;
    _runZonedWithOnDone(() => sink.add(response), () {
      trace.record(new UserCallbackDoneEvent());
      _openCallbacks--;
      if (_openCallbacks == 0 && _closed && !_sentDone) {
        trace.record(new RequestOverEvent());
        _sentDone = true;
      }
    }, trace);
  }

  void handleError(error, EventSink<Response> sink, Trace trace) {
    trace.record(new UserCallbackQueuedEvent());
    _openCallbacks++;
    _runZonedWithOnDone(() => sink.addError(error), () {
      trace.record(new UserCallbackDoneEvent());
      _openCallbacks--;
      if (_openCallbacks == 0 && _closed && !_sentDone) {
        trace.record(new RequestOverEvent());
        _sentDone = true;
      }
    }, trace);
  }

  void handleDone(EventSink<Response> sink, Trace trace) {
    _closed = true;
    if (_openCallbacks == 0 && !_sentDone) {
      trace.record(new RequestOverEvent());
      _sentDone = true;
    }
  }

  void handleCancel(Trace trace) {
    _closed = true;
    if (_openCallbacks == 0 && !_sentDone) {
      trace.record(new RequestOverEvent());
      _sentDone = true;
    }
  }

  static bool traceDonePredicate(TraceEvent event) => event is RequestOverEvent;
}

/// Fired when the user callback of a response is queued.
class UserCallbackQueuedEvent implements TraceEvent {
  factory UserCallbackQueuedEvent() => const UserCallbackQueuedEvent._private();

  const UserCallbackQueuedEvent._private();

  String toString() => 'streamy.userCallback.start';
}

/// Fired when the user callback of a response completes.
class UserCallbackDoneEvent implements TraceEvent {
  factory UserCallbackDoneEvent() => const UserCallbackDoneEvent._private();

  const UserCallbackDoneEvent._private();

  String toString() => 'streamy.userCallback.done';
}

/// Fired at the beginning of an asynchronous operation that happens during a user callback.
class UserCallbackAsyncEnterEvent implements TraceEvent {
  factory UserCallbackAsyncEnterEvent() => const UserCallbackAsyncEnterEvent._private();

  const UserCallbackAsyncEnterEvent._private();

  String toString() => 'streamy.userCallback.async.start';
}

// Fired at the end of an asynchronous operation that happens during a user callback.
class UserCallbackAsyncExitEvent implements TraceEvent {
  factory UserCallbackAsyncExitEvent() => const UserCallbackAsyncExitEvent._private();

  const UserCallbackAsyncExitEvent._private();

  String toString() => 'streamy.userCallback.async.done';
}<commit_msg>Fix comment.<commit_after>part of streamy.runtime;

/// An [EventTransformer] that de-duplicates entities. This will cause
/// metadata about the entity (Entity.streamy) to be inaccurate, but will
/// prevent multiple values from being published on [Stream]s when core [Entity]
/// data has not changed.
class EntityDedupTransformer extends EventTransformer {
  var _last = null;

  EntityDedupTransformer() : super();

  void handleData(Response<T> response, EventSink<T> sink, Trace trace) {
    if (!Entity.deepEquals(response.entity, _last)) {
      sink.add(response);
    }
    _last = response.entity;
  }
}

/// An [EventTransformer] that closes the stream after the RPC reply is
/// received.
class OneShotRequestTransformer extends EventTransformer {

  const OneShotRequestTransformer() : super();

  void handleData(Response response, EventSink<Response> sink, Trace trace) {
    sink.add(response);
    if (response.source == Source.RPC) {
      sink.close();
    }
  }
}

/// An [EventTransformer] that clones frozen entities, to make them mutable.
class MutableTransformer extends EventTransformer {

  const MutableTransformer() : super();

  void handleData(Response response, EventSink<Response> sink, Trace trace) {
    if (response.entity.isFrozen) {
      sink.add(new Response(response.entity.clone(), response.source, response.ts));
    } else {
      sink.add(response);
    }
  }
}

/// An [EventTransformer] that traces user callback timings. Should be the last
/// transformer before user code.
class UserCallbackTracingTransformer extends EventTransformer {

  var _openCallbacks = 0;
  var _closed = false;
  var _sentDone = false;

  UserCallbackTracingTransformer() : super();

  void handleData(Response response, EventSink<Response> sink, Trace trace) {
    trace.record(new UserCallbackQueuedEvent());
    _openCallbacks++;
    _runZonedWithOnDone(() => sink.add(response), () {
      trace.record(new UserCallbackDoneEvent());
      _openCallbacks--;
      if (_openCallbacks == 0 && _closed && !_sentDone) {
        trace.record(new RequestOverEvent());
        _sentDone = true;
      }
    }, trace);
  }

  void handleError(error, EventSink<Response> sink, Trace trace) {
    trace.record(new UserCallbackQueuedEvent());
    _openCallbacks++;
    _runZonedWithOnDone(() => sink.addError(error), () {
      trace.record(new UserCallbackDoneEvent());
      _openCallbacks--;
      if (_openCallbacks == 0 && _closed && !_sentDone) {
        trace.record(new RequestOverEvent());
        _sentDone = true;
      }
    }, trace);
  }

  void handleDone(EventSink<Response> sink, Trace trace) {
    _closed = true;
    if (_openCallbacks == 0 && !_sentDone) {
      trace.record(new RequestOverEvent());
      _sentDone = true;
    }
  }

  void handleCancel(Trace trace) {
    _closed = true;
    if (_openCallbacks == 0 && !_sentDone) {
      trace.record(new RequestOverEvent());
      _sentDone = true;
    }
  }

  static bool traceDonePredicate(TraceEvent event) => event is RequestOverEvent;
}

/// Fired when the user callback of a response is queued.
class UserCallbackQueuedEvent implements TraceEvent {
  factory UserCallbackQueuedEvent() => const UserCallbackQueuedEvent._private();

  const UserCallbackQueuedEvent._private();

  String toString() => 'streamy.userCallback.start';
}

/// Fired when the user callback of a response completes.
class UserCallbackDoneEvent implements TraceEvent {
  factory UserCallbackDoneEvent() => const UserCallbackDoneEvent._private();

  const UserCallbackDoneEvent._private();

  String toString() => 'streamy.userCallback.done';
}

/// Fired at the beginning of an asynchronous operation that happens during a user callback.
class UserCallbackAsyncEnterEvent implements TraceEvent {
  factory UserCallbackAsyncEnterEvent() => const UserCallbackAsyncEnterEvent._private();

  const UserCallbackAsyncEnterEvent._private();

  String toString() => 'streamy.userCallback.async.start';
}

/// Fired at the end of an asynchronous operation that happens during a user callback.
class UserCallbackAsyncExitEvent implements TraceEvent {
  factory UserCallbackAsyncExitEvent() => const UserCallbackAsyncExitEvent._private();

  const UserCallbackAsyncExitEvent._private();

  String toString() => 'streamy.userCallback.async.done';
}
{'fraction_non_alphanumeric': 0.1084029712163417, 'fraction_numerical': 0.001160631383472609, 'mean_word_length': 4.039181286549708, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 39, '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': '3683070', 'n_tokens_mistral': 2430, 'n_tokens_neox': 2386, 'n_words': 669}