Try to do a regulator, draft

regulator.js

'use strict';

/**
 * A regulator connects Source and Reducer/Effect with regards of responsiveness.
 */

class Regulator {

  constructor(configs) {
    this._source  = null;
    this._reducer = (acc, current) => { acc.push(current); return acc };
    this._effect  = null;

    // 60 FPS is the holy number (120 in the future, maybe).
    this._msSoft = 16;
    this._msHard = 16;
    if ('undefined' !== typeof configs) {
      if (configs.limitation) {
        this._msSoft  = configs.limitation.soft;
        this._msHard  = configs.limitation.hard;
      }
      this._requestAnimationFrame = configs.requestAnimationFrame || window.requestAnimationFrame.bind(window);
      this._calibrateStrategy = configs.calibrationStrategy || this._defaultCalibrate.bind(this);
    } else {
      // Will throw if it starts without customized frameRequester AND the default one in the browser
      this._requestAnimationFrame = window.requestAnimationFrame.bind(window);
      this._calibrateStrategy = this._defaultCalibrate.bind(this);
    }
    this._fetchFactor = 1;  // Start from 1 datum; increasing like TCP (the default strategy)
    this._accumulated = [];
    this._timestamp = this._timestampGenerator();
    this._timediff = this._timediffGenerator();
  }

  /**
   * |source|: must be a Generator. After calling it,
   * the result can be poll in the iterator way (|next()|).
   */
  source(source) {
    this._source = source();
    return this;
  }

  /**
   * reducer:: (acc, current) -> acc'
   *
   * From reducing a whole data list to reducing chunks.
   * If user doesn't call reduce to set it, means the data don't need tramsformation.
   */
  reducer(reducer, init) {
    this._reducer = reducer;
    this._accumulated = init;
    return this;
  }

  /**
   * effect:: (acc, current) -> Effect ()
   */
  effect(effect) {
    // TODO: what if effect is asynchronous?
    this._effect = effect;
    return this;
  }

  perform() {
    this._doPerform();
  }

  _doPerform() {
    // 1. request a frame
    // 2. mark |begin|
    // 3. get 1 datum
    // 4. perform |reducer|
    // 5. feed |effect|
    // 6. mark |end|
    // 7. get |diff|
    this._requestAnimationFrame(() => {
      let begin = this._timestamp();
      let [done, chunk] = this._fetch();
      if (!done) {
        this._accumulated = this._reducer(this._accumulated, chunk);
        this._effect(this._accumulated, chunk);
        let diff = this._timediff(begin);
        this._fetchFactor = this._calibrateStrategy(diff, this._fetchFactor,
                                                    this._msSoft, this._msHard,
                                                    this._raiseSoft.bind(this), this._raiseHard.bind(this), chunk.length);
        this._doPerform();
      }
    });
  }

  /**
   * Resemble TCP congestion flow.
   * |factor|: how much data will be fetched.
   */
  _defaultCalibrate(elapsed, factor, msSoft, msHard, raiseSoft, raiseHard, chunkSize) {
      if (elapsed < msSoft) {
        return factor << 1;
      } else if (elapsed > msSoft && elapsed < msHard) {
        raiseSoft(elapsed, msSoft, chunkSize);
        return 1;
      } else {
        raiseHard(elapsed, msHard, chunkSize);
        return 1;
      }
  }

  _raiseSoft(elapsed, limitation, chunkSize) {
    console.error(`Soft limitation exceeded: ${elapsed} > ${limitation}; with a ${chunkSize} size chunk`);
  }

  _raiseHard(elapsed, limitation, chunkSize) {
    console.error(`Hard limitation exceeded: ${elapsed} > ${limitation}; with a ${chunkSize} size chunk`);
  }

  _fetch() {
    let result = [ false, []];
    for (let i = 0; i < this._fetchFactor; i ++) {
      let {done, value} = this._source.next();
      if (done) {
        result[0] = true;
        break;
      } else {
        result[1].push(value);
      }
    }
    return result;
  }

  _timediffGenerator() {
    if ('undefined' !== typeof process && 'undefined' !== typeof process.hrtime) {
      return (begin) => {
        let [sec, nanosec] = process.hrtime(begin);
        return (sec * 1e9 + nanosec) / 1e6;   // to ms
      };
    } else if ('undefined' !== typeof performance && 'undefined' !== typeof performance.now) {
      return (begin) => performance.now() - begin;
    }
    throw new Error('Cannot find a way to generate timediff');
  }

  _timestampGenerator() {
    console.log(typeof process, typeof process.hrtime);
    if ('undefined' !== typeof process && 'undefined' !== typeof process.hrtime) {
      return process.hrtime.bind(process);
    } else if ('undefined' !== typeof performance && 'undefined' !== typeof performance.now) {
      return performance.now.bind(performance);
    }
    throw new Error('Cannot find a way to generate timestamp');
  }

}

// Make it as building interface.
exports.Regulator = function(configs) {
  return new Regulator(configs);
};