graphbrainz/lib/rate-limit.js

'use strict';

Object.defineProperty(exports, "__esModule", {
  value: true
});

var _toConsumableArray2 = require('babel-runtime/helpers/toConsumableArray');

var _toConsumableArray3 = _interopRequireDefault(_toConsumableArray2);

var _promise = require('babel-runtime/core-js/promise');

var _promise2 = _interopRequireDefault(_promise);

var _classCallCheck2 = require('babel-runtime/helpers/classCallCheck');

var _classCallCheck3 = _interopRequireDefault(_classCallCheck2);

var _createClass2 = require('babel-runtime/helpers/createClass');

var _createClass3 = _interopRequireDefault(_createClass2);

function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }

var debug = require('debug')('graphbrainz:rate-limit');

var RateLimit = function () {
  function RateLimit() {
    var _ref = arguments.length > 0 && arguments[0] !== undefined ? arguments[0] : {},
        _ref$limit = _ref.limit,
        limit = _ref$limit === undefined ? 1 : _ref$limit,
        _ref$period = _ref.period,
        period = _ref$period === undefined ? 1000 : _ref$period,
        _ref$concurrency = _ref.concurrency,
        concurrency = _ref$concurrency === undefined ? limit || 1 : _ref$concurrency,
        _ref$defaultPriority = _ref.defaultPriority,
        defaultPriority = _ref$defaultPriority === undefined ? 1 : _ref$defaultPriority;

    (0, _classCallCheck3.default)(this, RateLimit);

    this.limit = limit;
    this.period = period;
    this.defaultPriority = defaultPriority;
    this.concurrency = concurrency;
    this.queues = [];
    this.numPending = 0;
    this.periodStart = null;
    this.periodCapacity = this.limit;
    this.timer = null;
    this.pendingFlush = false;
    this.prevTaskID = null;
  }

  (0, _createClass3.default)(RateLimit, [{
    key: 'nextTaskID',
    value: function nextTaskID() {
      var prevTaskID = arguments.length > 0 && arguments[0] !== undefined ? arguments[0] : this.prevTaskID;

      var id = (prevTaskID || 0) + 1;
      this.prevTaskID = id;
      return id;
    }
  }, {
    key: 'enqueue',
    value: function enqueue(fn, args) {
      var _this = this;

      var priority = arguments.length > 2 && arguments[2] !== undefined ? arguments[2] : this.defaultPriority;

      priority = Math.max(0, priority);
      return new _promise2.default(function (resolve, reject) {
        var queue = _this.queues[priority] = _this.queues[priority] || [];
        var id = _this.nextTaskID();
        debug('Enqueuing task. id=' + id + ' priority=' + priority);
        queue.push({ fn: fn, args: args, resolve: resolve, reject: reject, id: id });
        if (!_this.pendingFlush) {
          _this.pendingFlush = true;
          process.nextTick(function () {
            _this.pendingFlush = false;
            _this.flush();
          });
        }
      });
    }
  }, {
    key: 'dequeue',
    value: function dequeue() {
      var task = void 0;
      for (var i = this.queues.length - 1; i >= 0; i--) {
        var queue = this.queues[i];
        if (queue && queue.length) {
          task = queue.shift();
        }
        if (!queue || !queue.length) {
          this.queues.length = i;
        }
        if (task) {
          break;
        }
      }
      return task;
    }
  }, {
    key: 'flush',
    value: function flush() {
      var _this2 = this;

      if (this.numPending < this.concurrency && this.periodCapacity > 0) {
        var task = this.dequeue();
        if (task) {
          var resolve = task.resolve,
              reject = task.reject,
              fn = task.fn,
              args = task.args,
              id = task.id;

          if (this.timer == null) {
            var now = Date.now();
            var timeout = this.period;
            if (this.periodStart != null) {
              var delay = now - (this.periodStart + timeout);
              if (delay > 0 && delay <= timeout) {
                timeout -= delay;
              }
            }
            this.periodStart = now;
            this.timer = setTimeout(function () {
              _this2.timer = null;
              _this2.periodCapacity = _this2.limit;
              _this2.flush();
            }, timeout);
          }
          this.numPending += 1;
          this.periodCapacity -= 1;
          var onResolve = function onResolve(value) {
            _this2.numPending -= 1;
            resolve(value);
            _this2.flush();
          };
          var onReject = function onReject(err) {
            _this2.numPending -= 1;
            reject(err);
            _this2.flush();
          };
          debug('Running task. id=' + id);
          fn.apply(undefined, (0, _toConsumableArray3.default)(args)).then(onResolve, onReject);
          this.flush();
        }
      }
    }
  }]);
  return RateLimit;
}();

exports.default = RateLimit;