<commit_before>'use strict';

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


var channelReducer = function(ch) {
  return {
    init: function() {
      return true;
    },
    result: function(open) {
      return open;
    },
    step: function(open, val) {
      return core.go(function*() {
        var success;

        open = yield open;
        val = yield val;
        if (val !== undefined)
          return yield ch.push(val);
        else
          return open;
      });
    }
  };
};


module.exports = function(xform, buf, isReduced, deref) {
  var ch = chan.chan(buf);
  var xf = xform(channelReducer(ch));
  var open = xf.init();

  return {
    push: function(val, handler) {
      return core.go(function*() {
        open = yield xf.step(open, val);
        if (isReduced(open)) {
          yield xf.result(yield deref(open));
          open = false;
          ch.close();
        };
        return open;
      });
    },

    pull: function(handler) {
      return ch.pull(handler);
    },

    close: function() {
      core.go(function*() {
        yield xf.result(open);
        open = false;
        ch.close();
      });
    }
  };
};<commit_msg>further work on the tachan code<commit_after>'use strict';

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


var channelReducer = function(ch) {
  return {
    init: function() {
      return true;
    },
    result: function(t) {
      return t;
    },
    step: function(t, val) {
      return core.go(function*() {
        t = yield t;
        val = yield val;
        if (val !== undefined)
          return yield ch.push(val);
      });
    }
  };
};


module.exports = function(xform, buf, isReduced, deref) {
  var ch = chan.chan(buf);
  var open = true;
  var xf = xform(channelReducer(ch));
  xf.init();

  return {
    push: function(val, handler) {
      return core.go(function*() {
        if (open) {
          var result = yield xf.step(null, val);
          if (isReduced(result)) {
            yield deref(yield xf.result(yield deref(result)));
            open = false;
            ch.close();
          };
          return true;
        } else
          return false;
      });
    },

    pull: function(handler) {
      return ch.pull(handler);
    },

    close: function() {
      core.top(core.go(function*() {
        yield deref(yield xf.result());
        open = false;
        ch.close();
      }));
    }
  };
};
{'fraction_non_alphanumeric': 0.15155228758169934, 'fraction_numerical': 0.0, 'mean_word_length': 1.8476744186046512, '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': '5359025', 'n_tokens_mistral': 796, 'n_tokens_neox': 771, 'n_words': 185}