<commit_before>part of frappe.transformers;

class Zip<A, B, R> implements StreamTransformer<A, R> {
  final Stream<B> _other;
  final Combiner<A, B, R> _combiner;

  Zip(Stream<B> other, Combiner<A, B, R> combiner) :
    _other = other,
    _combiner = combiner;

  Stream<R> bind(Stream<A> stream) {
    StreamController<R> controller;
    StreamSubscription<A> subscriptionA;
    StreamSubscription<B> subscriptionB;

    void done() {
      subscriptionA.cancel();
      subscriptionB.cancel();
      controller.close();
    }

    controller = _createControllerLikeStream(stream: stream, onListen: () {
      var bufferA = new Queue();
      var bufferB = new Queue();

      void zipIfValuesExist() {
        if (bufferA.isNotEmpty && bufferB.isNotEmpty) {
          controller.add(_combiner(bufferA.removeFirst(), bufferB.removeFirst()));
        }
      }

      subscriptionA = stream.listen((value) {
        bufferA.addLast(value);
        zipIfValuesExist();
      }, onError: controller.addError, onDone: () => done());
      subscriptionB = _other.listen((value) {
        bufferB.addLast(value);
        zipIfValuesExist();
      }, onError: controller.addError, onDone: () => done());
    }, onCancel: () => done());

    return controller.stream;
  }
}<commit_msg>FPify the zip transformer<commit_after>part of frappe.transformers;

class Zip<A, B, R> implements StreamTransformer<A, R> {
  final Stream<B> _other;
  final Combiner<A, B, R> _combiner;

  Zip(Stream<B> other, Combiner<A, B, R> combiner) :
    _other = other,
    _combiner = combiner;

  Stream<R> bind(Stream<A> stream) {
    Queue appendToQueue(Queue queue, element) => queue..add(element);

    var bufferA = stream.transform(new Scan(new Queue<A>(), appendToQueue));
    var bufferB = _other.transform(new Scan(new Queue<B>(), appendToQueue));

    var combined = Combine.all([bufferA, bufferB]) as Stream<List<Queue>>;

    return combined
        .where((queues) => queues.first.isNotEmpty && queues.last.isNotEmpty)
        .map((queues) => _combiner(queues.first.removeFirst(), queues.last.removeFirst()));
  }
}
{'fraction_non_alphanumeric': 0.14211026615969582, 'fraction_numerical': 0.0, 'mean_word_length': 3.5268817204301075, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 32, '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': '7222624', 'n_tokens_mistral': 688, 'n_tokens_neox': 677, 'n_words': 146}