<commit_before>'use strict';

var go = require('./core').go;
var cc = require('./channels');

exports.source = function*(gen, outch, done) {
  for (;;) {
    var step = gen.next();
    if (step.done || !(yield cc.push(outch, step.value)))
      break;
  }
  yield cc.push(done, true);
};

exports.map = function*(fn, inch, outch, done) {
  var val;
  while((val = yield cc.pull(inch)) !== undefined)
    if(!(yield cc.push(outch, fn(val))))
      break;
  yield cc.push(done, true);
};

exports.filter = function*(pred, inch, outch, done) {
  var val;
  while((val = yield cc.pull(inch)) !== undefined)
    if (pred(val))
      if (!(yield cc.push(outch, val)))
        break;
  yield cc.push(done, true);
};

exports.reductions = function*(fn, inch, outch, done) {
  var val, acc;
  while ((val = yield cc.pull(inch)) !== undefined) {
    if (acc === undefined)
      acc = val;
    else
      acc = fn(acc, val);
    if (!(yield cc.push(outch, acc)))
      break;
  }
  yield cc.push(done, true);
};

exports.take = function*(n, inch, outch, done) {
  var val, i;
  for (i = 0; i < n; ++i) {
    val = yield cc.pull(inch);
    if (val === undefined || !(yield cc.push(outch, val)))
      break;
  }
  yield cc.push(done, true);
};

exports.takeWithTimeout = function*(ms, inch, outch, done) {
  var t = cc.timeout(ms);
  var val;
  while((val = (yield cc.select(t, inch)).value) !== undefined)
    if (!(yield cc.push(outch, val)))
      break;
  yield cc.push(done, true);
};

var rest = function(taker) {
  return function*(arg, inch, outch, done) {
    var sink = cc.chan(0);
    var sunk = cc.chan();
    go(taker, arg, inch, sink, sunk);
    yield cc.pull(sunk);
    go(exports.map, function(x) { return x; }, inch, outch, done);
  };
};

exports.drop = rest(exports.take);
exports.dropWithTimeout = rest(exports.takeWithTimeout);

exports.takeUntil = function*(pred, inch, outch, done) {
  var val;
  while((val = yield cc.pull(inch)) !== undefined)
    if (!(yield cc.push(outch, val)) || pred(val))
      break;
  yield cc.push(done, true);
};

exports.dropWhile = function*(pred, inch, outch, done) {
  var val;
  var go = false;
  while((val = yield cc.pull(inch)) !== undefined) {
    go = go || !pred(val);
    if (go && !(yield cc.push(outch, val)))
      break;
  }
  yield cc.push(done, true);
};

exports.merge = function*(inchs, outch, done) {
  var active = inchs.slice();

  while (active.length > 0) {
    var res = yield cc.select.apply(null, active);
    if (res.value === undefined)
      active.splice(res.index, 1);
    else
      if (!(yield cc.push(outch, res.value)))
        break;
  }
  yield cc.push(done, true);
};

exports.combine = function*(inchs, outch, done) {
  var results, active, indices, i, res;

  active  = inchs.slice();
  indices = []
  for (i = 0; i < inchs.length; ++i)
    indices.push(i);

  results = new Array(inchs.length);

  while (active.length > 0) {
    res = yield cc.select.apply(null, active);
    i = res.index;

    if (res.value === undefined) {
      active.splice(i, 1);
      indices.splice(i, 1);
    } else {
      results[indices[i]] = res.value;
      if (!(yield cc.push(outch, results.slice())))
        break;
    }
  }
  yield cc.push(done, true);
};

exports.zip = function*(inchs, outch, done) {
  var merged = cc.chan();
  var n = inchs.length;
  var val = new Array(n);
  var i, terminate;

  var read = function(i) {
    go(function*() {
      yield cc.push(merged, [i, yield cc.pull(inchs[i])]);
    });
  };

  while (true) {
    for (i = 0; i < n; ++ i)
      read(i);<commit_msg>rewrite merge() and combine() not to select over input channels<commit_after>'use strict';

var go = require('./core').go;
var cc = require('./channels');

exports.source = function*(gen, outch, done) {
  for (;;) {
    var step = gen.next();
    if (step.done || !(yield cc.push(outch, step.value)))
      break;
  }
  yield cc.push(done, true);
};

exports.map = function*(fn, inch, outch, done) {
  var val;
  while((val = yield cc.pull(inch)) !== undefined)
    if(!(yield cc.push(outch, fn(val))))
      break;
  yield cc.push(done, true);
};

exports.filter = function*(pred, inch, outch, done) {
  var val;
  while((val = yield cc.pull(inch)) !== undefined)
    if (pred(val))
      if (!(yield cc.push(outch, val)))
        break;
  yield cc.push(done, true);
};

exports.reductions = function*(fn, inch, outch, done) {
  var val, acc;
  while ((val = yield cc.pull(inch)) !== undefined) {
    if (acc === undefined)
      acc = val;
    else
      acc = fn(acc, val);
    if (!(yield cc.push(outch, acc)))
      break;
  }
  yield cc.push(done, true);
};

exports.take = function*(n, inch, outch, done) {
  var val, i;
  for (i = 0; i < n; ++i) {
    val = yield cc.pull(inch);
    if (val === undefined || !(yield cc.push(outch, val)))
      break;
  }
  yield cc.push(done, true);
};

exports.takeWithTimeout = function*(ms, inch, outch, done) {
  var t = cc.timeout(ms);
  var val;
  while((val = (yield cc.select(t, inch)).value) !== undefined)
    if (!(yield cc.push(outch, val)))
      break;
  yield cc.push(done, true);
};

var rest = function(taker) {
  return function*(arg, inch, outch, done) {
    var sink = cc.chan(0);
    var sunk = cc.chan();
    go(taker, arg, inch, sink, sunk);
    yield cc.pull(sunk);
    go(exports.map, function(x) { return x; }, inch, outch, done);
  };
};

exports.drop = rest(exports.take);
exports.dropWithTimeout = rest(exports.takeWithTimeout);

exports.takeUntil = function*(pred, inch, outch, done) {
  var val;
  while((val = yield cc.pull(inch)) !== undefined)
    if (!(yield cc.push(outch, val)) || pred(val))
      break;
  yield cc.push(done, true);
};

exports.dropWhile = function*(pred, inch, outch, done) {
  var val;
  var go = false;
  while((val = yield cc.pull(inch)) !== undefined) {
    go = go || !pred(val);
    if (go && !(yield cc.push(outch, val)))
      break;
  }
  yield cc.push(done, true);
};

exports.merge = function*(inchs, outch, done) {
  var active = cc.chan();

  inchs.forEach(function(ch) {
    go(function*() {
      var val;
      while((val = (yield cc.select(active, ch)).value) !== undefined)
        if (!(yield cc.push(outch, val)))
          break;
      active.close();
    });
  });

  yield cc.pull(active);
  yield cc.push(done, true);
};

exports.combine = function*(inchs, outch, done) {
  var result = new Array(inchs.length);
  var active = cc.chan();

  var run = function(i) {
    go(function*() {
      var val;
      while((val = (yield cc.select(active, inchs[i])).value) !== undefined) {
        result[i] = val;
        if (!(yield cc.push(outch, result.slice())))
          break;
      }
      active.close();
    });
  };

  for (var i = 0; i < inchs.length; ++i)
    run(i);

  yield cc.pull(active);
  yield cc.push(done, true);
};

exports.zip = function*(inchs, outch, done) {
  var merged = cc.chan();
  var n = inchs.length;
  var val = new Array(n);
  var i, terminate;

  var read = function(i) {
    go(function*() {
      yield cc.push(merged, [i, yield cc.pull(inchs[i])]);
    });
  };

  while (true) {
    for (i = 0; i < n; ++ i)
      read(i);
{'fraction_non_alphanumeric': 0.18874264911789415, 'fraction_numerical': 0.0018202184262111453, 'mean_word_length': 2.9182665935271532, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, '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': '3580176', 'n_tokens_mistral': 2628, 'n_tokens_neox': 2578, 'n_words': 671}