<commit_before>'use strict';

module.exports = (api) => {

  const collectorMethodNames = ['pick', 'fail', 'take', 'timeout', 'done'];

  api.metasync.collect = (
    expected // count or array of string
    // returns: collector functor
  ) => {
    const isCount = typeof(expected) === 'number';
    const isKeys = Array.isArray(expected);
    if (!(isCount || isKeys)) throw new TypeError('Unexpected type');
    let keys = null;
    if (isKeys) {
      keys = new Set(expected);
      expected = expected.length;
    }
    let count = 0;
    let takeCbCount = 0;
    let timer = null;
    let done = null;
    let isDistinct = false;
    let isDone = false;
    const data = {};
    const returnIfDone = (fn) => (...args) => {
      if (!isDone) return fn(...args);
    };

    const collector = returnIfDone((key, err, value) => {
      if (!isDistinct || !(key in data)) {
        if (!isCount && !keys.has(key)) return;
        count++;
      }
      if (err) return done(err, data);
      data[key] = value;
      if (expected === count) {
        if (timer) clearTimeout(timer);
        done(null, data);
      }
      return collector;
    });

    collector.pick = (key, value) => collector(key, null, value);

    collector.fail = (key, err) => collector(key, err);

    collector.take = (key, fn, ...args) => {
      if (count + takeCbCount < expected) {
        takeCbCount++;
        fn(...args, (...cbArgs) => {
          takeCbCount--;
          collector(key, ...cbArgs);
        });
      }
      return collector;
    };

    collector.timeout = (msec) => {
      if (msec) {
        timer = setTimeout(() => {
          const err = new Error('Collector timeout');
          done(err, data);
        }, msec);
        timer.unref();
      }
      return collector;
    };

    collector.done = (cb) => {
      done = api.common.cb((...args) => {
        isDone = true;
        let methodName;
        for (methodName of collectorMethodNames) {
          collector[methodName] = api.common.emptiness;
        }
        cb(...args);
      });
      return collector;
    };

    collector.distinct = () => {
      isDistinct = true;
      return collector;
    };

    let methodName;
    for (methodName of collectorMethodNames) {
      collector[methodName] = returnIfDone(collector[methodName]);
    }

    return collector;
  };

  function Collector() {}

  Collector.prototype.on = function(
    // Collector events:
    eventName,
    listener // handler function
    // on('error', function(err, key))
    // on('timeout', function(err, data))
    // on('done', function(errs, data))
  ) {
    if (eventName in this.events) {
      this.events[eventName] = listener;
    }
  };

  Collector.prototype.emit = function(
    // Emit Collector events
    eventName, err, data
  ) {
    const event = this.events[eventName];
    if (event) event(err, data);
  };

  api.metasync.DataCollector = function(
    expected, // number of collect() calls expected
    timeout // collect timeout (optional)
    // returns: instance of DataCollector
  ) {
    this.expected = expected;
    this.timeout = timeout;
    this.count = 0;
    this.data = {};
    this.errs = [];
    this.events = {
      error: null,
      timeout: null,
      done: null
    };
    const collector = this;
    if (this.timeout) {
      this.timer = setTimeout(() => {
        const err = new Error('DataCollector timeout');
        collector.emit('timeout', err, collector.data);
      }, timeout);
    }
  };

  api.util.inherits(api.metasync.DataCollector, Collector);

  api.metasync.DataCollector.prototype.collect = function(
    // Push data to collector
    key, // key in result data
    data // value or error instance
  ) {
    this.count++;
    if (data instanceof Error) {
      this.errs[key] = data;
      this.emit('error', data, key);
    } else {
      this.data[key] = data;
    }
    if (this.expected === this.count) {
      if (this.timer) clearTimeout(this.timer);
      const errs = this.errs.length ? this.errs : null;
      this.emit('done', errs, this.data);
    }
  };

  api.metasync.KeyCollector = function(
    // Key Collector
    keys, // array of keys, example: ['config', 'users', 'cities']
    timeout // collect timeout (optional)
    // returns: instance of DataCollector
  ) {
    this.isDone = false;
    this.keys = keys;
    this.expected = keys.length;
    this.count = 0;
    this.timeout = timeout;
    this.data = {};
    this.errs = [];
    this.events = {
      error: null,
      timeout: null,
      done: null
    };
    const collector = this;
    if (this.timeout) {
      this.timer = setTimeout(() => {
        const err = new Error('KeyCollector timeout');
        collector.emit('timeout', err, collector.data);
      }, timeout);
    }
  };

  api.util.inherits(api.metasync.KeyCollector, Collector);

  api.metasync.KeyCollector.prototype.collect = function(
    key,
    data
  ) {
    if (this.keys.includes(key)) {
      this.count++;
      if (data instanceof Error) {
        this.errs[key] = data;
        this.emit('error', data, key);
      } else {
        this.data[key] = data;
      }
      if (this.expected === this.count) {
        if (this.timer) clearTimeout(this.timer);
        const errs = this.errs.length ? this.errs : null;
        this.emit('done', errs, this.data);
      }
    }
  };

  api.metasync.KeyCollector.prototype.stop = function() {
  };

  api.metasync.KeyCollector.prototype.pause = function() {
  };

  api.metasync.KeyCollector.prototype.resume = function() {
  };

};
<commit_msg>Change `distinct` for collector<commit_after>'use strict';

module.exports = (api) => {

  const collectorMethodNames = ['pick', 'fail', 'take', 'timeout', 'done'];

  api.metasync.collect = (
    expected // count or array of string
    // returns: collector functor
  ) => {
    const isCount = typeof(expected) === 'number';
    const isKeys = Array.isArray(expected);
    if (!(isCount || isKeys)) throw new TypeError('Unexpected type');
    let keys = null;
    if (isKeys) {
      keys = new Set(expected);
      expected = expected.length;
    }
    let count = 0;
    let takeCbCount = 0;
    let timer = null;
    let done = null;
    let isDistinct = false;
    let isDone = false;
    const data = {};
    const returnIfDone = (fn) => (...args) => {
      if (!isDone) return fn(...args);
    };

    const collector = returnIfDone((key, err, value) => {
      if (!isDistinct || !(key in data)) {
        if (!isCount && !keys.has(key)) return;
        count++;
      }
      if (err) return done(err, data);
      data[key] = value;
      if (expected === count) {
        if (timer) clearTimeout(timer);
        done(null, data);
      }
      return collector;
    });

    collector.pick = (key, value) => collector(key, null, value);

    collector.fail = (key, err) => collector(key, err);

    collector.take = (key, fn, ...args) => {
      if (count + takeCbCount < expected) {
        takeCbCount++;
        fn(...args, (...cbArgs) => {
          takeCbCount--;
          collector(key, ...cbArgs);
        });
      }
      return collector;
    };

    collector.timeout = (msec) => {
      if (msec) {
        timer = setTimeout(() => {
          const err = new Error('Collector timeout');
          done(err, data);
        }, msec);
        timer.unref();
      }
      return collector;
    };

    collector.done = (cb) => {
      done = api.common.cb((...args) => {
        isDone = true;
        let methodName;
        for (methodName of collectorMethodNames) {
          collector[methodName] = api.common.emptiness;
        }
        cb(...args);
      });
      return collector;
    };

    collector.distinct = (value = true) => {
      isDistinct = value;
      return collector;
    };

    let methodName;
    for (methodName of collectorMethodNames) {
      collector[methodName] = returnIfDone(collector[methodName]);
    }

    return collector;
  };

  function Collector() {}

  Collector.prototype.on = function(
    // Collector events:
    eventName,
    listener // handler function
    // on('error', function(err, key))
    // on('timeout', function(err, data))
    // on('done', function(errs, data))
  ) {
    if (eventName in this.events) {
      this.events[eventName] = listener;
    }
  };

  Collector.prototype.emit = function(
    // Emit Collector events
    eventName, err, data
  ) {
    const event = this.events[eventName];
    if (event) event(err, data);
  };

  api.metasync.DataCollector = function(
    expected, // number of collect() calls expected
    timeout // collect timeout (optional)
    // returns: instance of DataCollector
  ) {
    this.expected = expected;
    this.timeout = timeout;
    this.count = 0;
    this.data = {};
    this.errs = [];
    this.events = {
      error: null,
      timeout: null,
      done: null
    };
    const collector = this;
    if (this.timeout) {
      this.timer = setTimeout(() => {
        const err = new Error('DataCollector timeout');
        collector.emit('timeout', err, collector.data);
      }, timeout);
    }
  };

  api.util.inherits(api.metasync.DataCollector, Collector);

  api.metasync.DataCollector.prototype.collect = function(
    // Push data to collector
    key, // key in result data
    data // value or error instance
  ) {
    this.count++;
    if (data instanceof Error) {
      this.errs[key] = data;
      this.emit('error', data, key);
    } else {
      this.data[key] = data;
    }
    if (this.expected === this.count) {
      if (this.timer) clearTimeout(this.timer);
      const errs = this.errs.length ? this.errs : null;
      this.emit('done', errs, this.data);
    }
  };

  api.metasync.KeyCollector = function(
    // Key Collector
    keys, // array of keys, example: ['config', 'users', 'cities']
    timeout // collect timeout (optional)
    // returns: instance of DataCollector
  ) {
    this.isDone = false;
    this.keys = keys;
    this.expected = keys.length;
    this.count = 0;
    this.timeout = timeout;
    this.data = {};
    this.errs = [];
    this.events = {
      error: null,
      timeout: null,
      done: null
    };
    const collector = this;
    if (this.timeout) {
      this.timer = setTimeout(() => {
        const err = new Error('KeyCollector timeout');
        collector.emit('timeout', err, collector.data);
      }, timeout);
    }
  };

  api.util.inherits(api.metasync.KeyCollector, Collector);

  api.metasync.KeyCollector.prototype.collect = function(
    key,
    data
  ) {
    if (this.keys.includes(key)) {
      this.count++;
      if (data instanceof Error) {
        this.errs[key] = data;
        this.emit('error', data, key);
      } else {
        this.data[key] = data;
      }
      if (this.expected === this.count) {
        if (this.timer) clearTimeout(this.timer);
        const errs = this.errs.length ? this.errs : null;
        this.emit('done', errs, this.data);
      }
    }
  };

  api.metasync.KeyCollector.prototype.stop = function() {
  };

  api.metasync.KeyCollector.prototype.pause = function() {
  };

  api.metasync.KeyCollector.prototype.resume = function() {
  };

};

{'fraction_non_alphanumeric': 0.15045198245771055, 'fraction_numerical': 0.0007160118141949343, 'mean_word_length': 2.196224256292906, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 35, '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': '4287245', 'n_tokens_mistral': 3562, 'n_tokens_neox': 3478, 'n_words': 823}