<commit_before>'use strict';

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


exports.timeout = function(ms) {
  var ch = channels.chan();
  var t = setTimeout(function() {
    clearTimeout(t);
    channels.close(ch);
  }, ms);
  return ch;
};


exports.ticker = function(ms) {
  var ch = channels.chan();
  var t;
  var step = function() {
    clearTimeout(t);
    t = setTimeout(step, ms);
    core.go(function*() {
      if (!(yield channels.push(ch, null)))
        clearTimeout(t);
    });
  };
  t = setTimeout(step, ms);
  return ch;
};


exports.each = function(fn, input) {
  return core.go(function*() {
    var val;
    while (undefined !== (val = yield channels.pull(input)))
      if (fn)
        yield fn(val);
  });
};


exports.pipe = function(input, output, keepOpen) {
  return core.go(function*() {
    var val;
    while (undefined !== (val = yield input.pull()))
      if (!(yield output.push(val)))
        break;
    if (!keepOpen) {
      input.close();
      output.close();
    }
  });
};


exports.createLock = function() {
  var _busy = channels.chan();
  _busy.push(null);

  return {
    acquire: function() { return _busy.pull(); },
    release: function() { return _busy.push(null); }
  };
};


exports.fromGenerator = function(gen) {
  var output = channels.chan();

  core.go(function*() {
    var step;
    while (true) {
      step = gen.next();
      if (step.done || !(yield output.push(step.value)))
        break
    }
    output.close();
  });

  return output;
};


exports.fromStream = function(stream, closeStream)
{<commit_msg>make fromGenerator() return a simplified custom channel<commit_after>'use strict';

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


exports.timeout = function(ms) {
  var ch = channels.chan();
  var t = setTimeout(function() {
    clearTimeout(t);
    channels.close(ch);
  }, ms);
  return ch;
};


exports.ticker = function(ms) {
  var ch = channels.chan();
  var t;
  var step = function() {
    clearTimeout(t);
    t = setTimeout(step, ms);
    core.go(function*() {
      if (!(yield channels.push(ch, null)))
        clearTimeout(t);
    });
  };
  t = setTimeout(step, ms);
  return ch;
};


exports.each = function(fn, input) {
  return core.go(function*() {
    var val;
    while (undefined !== (val = yield channels.pull(input)))
      if (fn)
        yield fn(val);
  });
};


exports.pipe = function(input, output, keepOpen) {
  return core.go(function*() {
    var val;
    while (undefined !== (val = yield input.pull()))
      if (!(yield output.push(val)))
        break;
    if (!keepOpen) {
      input.close();
      output.close();
    }
  });
};


exports.createLock = function() {
  var _busy = channels.chan();
  _busy.push(null);

  return {
    acquire: function() { return _busy.pull(); },
    release: function() { return _busy.push(null); }
  };
};


exports.fromGenerator = function(gen) {
  var _closed = false;

  return {
    pull: function(client) {
      var handler = client || core.defer();
      var value = undefined;

      if (!_closed) {
        var step = gen.next();
        if (step.done)
          _closed = true;
        else
          value = step.value;
      }

      handler.resolve(value);
      return handler;
    },
    close: function() {
      _closed = true;
    }
  };
};


exports.fromStream = function(stream, closeStream)
{
{'fraction_non_alphanumeric': 0.16920821114369503, 'fraction_numerical': 0.0, 'mean_word_length': 2.6248671625929862, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '3673819', 'n_tokens_mistral': 1122, 'n_tokens_neox': 1103, 'n_words': 262}