JezK
Edit File: blocking-queue.js
"use strict"; var _interopRequireDefault = require("@babel/runtime/helpers/interopRequireDefault"); Object.defineProperty(exports, "__esModule", { value: true }); exports["default"] = void 0; function _classCallCheck2() { var data = _interopRequireDefault(require("@babel/runtime/helpers/classCallCheck")); _classCallCheck2 = function _classCallCheck2() { return data; }; return data; } function _createClass2() { var data = _interopRequireDefault(require("@babel/runtime/helpers/createClass")); _createClass2 = function _createClass2() { return data; }; return data; } function _defineProperty2() { var data = _interopRequireDefault(require("@babel/runtime/helpers/defineProperty")); _defineProperty2 = function _defineProperty2() { return data; }; return data; } function _map() { var data = _interopRequireDefault(require("./map.js")); _map = function _map() { return data; }; return data; } var debug = require('debug')('yarn'); var BlockingQueue = /*#__PURE__*/function () { function BlockingQueue(alias) { var maxConcurrency = arguments.length > 1 && arguments[1] !== undefined ? arguments[1] : Infinity; (0, _classCallCheck2()["default"])(this, BlockingQueue); (0, _defineProperty2()["default"])(this, "concurrencyQueue", void 0); (0, _defineProperty2()["default"])(this, "warnedStuck", void 0); (0, _defineProperty2()["default"])(this, "maxConcurrency", void 0); (0, _defineProperty2()["default"])(this, "runningCount", void 0); (0, _defineProperty2()["default"])(this, "stuckTimer", void 0); (0, _defineProperty2()["default"])(this, "alias", void 0); (0, _defineProperty2()["default"])(this, "first", void 0); (0, _defineProperty2()["default"])(this, "queue", void 0); (0, _defineProperty2()["default"])(this, "running", void 0); this.concurrencyQueue = []; this.maxConcurrency = maxConcurrency; this.runningCount = 0; this.warnedStuck = false; this.alias = alias; this.first = true; this.running = (0, _map()["default"])(); this.queue = (0, _map()["default"])(); this.stuckTick = this.stuckTick.bind(this); } (0, _createClass2()["default"])(BlockingQueue, [{ key: "stillActive", value: function stillActive() { if (this.stuckTimer) { clearTimeout(this.stuckTimer); } this.stuckTimer = setTimeout(this.stuckTick, 5000); // We need to check the existence of unref because of https://github.com/facebook/jest/issues/4559 // $FlowFixMe: Node's setInterval returns a Timeout, not a Number this.stuckTimer.unref && this.stuckTimer.unref(); } }, { key: "stuckTick", value: function stuckTick() { if (this.runningCount === 1) { this.warnedStuck = true; debug("The ".concat(JSON.stringify(this.alias), " blocking queue may be stuck. 5 seconds ") + "without any activity with 1 worker: ".concat(Object.keys(this.running)[0])); } } }, { key: "push", value: function push(key, factory) { var _this = this; if (this.first) { this.first = false; } else { this.stillActive(); } return new Promise(function (resolve, reject) { // we're already running so push ourselves to the queue var queue = _this.queue[key] = _this.queue[key] || []; queue.push({ factory: factory, resolve: resolve, reject: reject }); if (!_this.running[key]) { _this.shift(key); } }); } }, { key: "shift", value: function shift(key) { var _this2 = this; if (this.running[key]) { delete this.running[key]; this.runningCount--; if (this.stuckTimer) { clearTimeout(this.stuckTimer); this.stuckTimer = null; } if (this.warnedStuck) { this.warnedStuck = false; debug("".concat(JSON.stringify(this.alias), " blocking queue finally resolved. Nothing to worry about.")); } } var queue = this.queue[key]; if (!queue) { return; } var _queue$shift = queue.shift(), resolve = _queue$shift.resolve, reject = _queue$shift.reject, factory = _queue$shift.factory; if (!queue.length) { delete this.queue[key]; } var next = function next() { _this2.shift(key); _this2.shiftConcurrencyQueue(); }; var run = function run() { _this2.running[key] = true; _this2.runningCount++; factory().then(function (val) { resolve(val); next(); return null; })["catch"](function (err) { reject(err); next(); }); }; this.maybePushConcurrencyQueue(run); } }, { key: "maybePushConcurrencyQueue", value: function maybePushConcurrencyQueue(run) { if (this.runningCount < this.maxConcurrency) { run(); } else { this.concurrencyQueue.push(run); } } }, { key: "shiftConcurrencyQueue", value: function shiftConcurrencyQueue() { if (this.runningCount < this.maxConcurrency) { var fn = this.concurrencyQueue.shift(); if (fn) { fn(); } } } }]); return BlockingQueue; }(); exports["default"] = BlockingQueue; //# sourceMappingURL=blocking-queue.js.map