280 lines
9.9 KiB
JavaScript
280 lines
9.9 KiB
JavaScript
|
"use strict";
|
||
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
||
|
|
const EventEmitter = require("eventemitter3");
|
||
|
|
const p_timeout_1 = require("p-timeout");
|
||
|
|
const priority_queue_1 = require("./priority-queue");
|
||
|
|
// eslint-disable-next-line @typescript-eslint/no-empty-function
|
||
|
|
const empty = () => { };
|
||
|
|
const timeoutError = new p_timeout_1.TimeoutError();
|
||
|
|
/**
|
||
|
|
Promise queue with concurrency control.
|
||
|
|
*/
|
||
|
|
class PQueue extends EventEmitter {
|
||
|
|
constructor(options) {
|
||
|
|
var _a, _b, _c, _d;
|
||
|
|
super();
|
||
|
|
this._intervalCount = 0;
|
||
|
|
this._intervalEnd = 0;
|
||
|
|
this._pendingCount = 0;
|
||
|
|
this._resolveEmpty = empty;
|
||
|
|
this._resolveIdle = empty;
|
||
|
|
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions
|
||
|
|
options = Object.assign({ carryoverConcurrencyCount: false, intervalCap: Infinity, interval: 0, concurrency: Infinity, autoStart: true, queueClass: priority_queue_1.default }, options);
|
||
|
|
if (!(typeof options.intervalCap === 'number' && options.intervalCap >= 1)) {
|
||
|
|
throw new TypeError(`Expected \`intervalCap\` to be a number from 1 and up, got \`${(_b = (_a = options.intervalCap) === null || _a === void 0 ? void 0 : _a.toString()) !== null && _b !== void 0 ? _b : ''}\` (${typeof options.intervalCap})`);
|
||
|
|
}
|
||
|
|
if (options.interval === undefined || !(Number.isFinite(options.interval) && options.interval >= 0)) {
|
||
|
|
throw new TypeError(`Expected \`interval\` to be a finite number >= 0, got \`${(_d = (_c = options.interval) === null || _c === void 0 ? void 0 : _c.toString()) !== null && _d !== void 0 ? _d : ''}\` (${typeof options.interval})`);
|
||
|
|
}
|
||
|
|
this._carryoverConcurrencyCount = options.carryoverConcurrencyCount;
|
||
|
|
this._isIntervalIgnored = options.intervalCap === Infinity || options.interval === 0;
|
||
|
|
this._intervalCap = options.intervalCap;
|
||
|
|
this._interval = options.interval;
|
||
|
|
this._queue = new options.queueClass();
|
||
|
|
this._queueClass = options.queueClass;
|
||
|
|
this.concurrency = options.concurrency;
|
||
|
|
this._timeout = options.timeout;
|
||
|
|
this._throwOnTimeout = options.throwOnTimeout === true;
|
||
|
|
this._isPaused = options.autoStart === false;
|
||
|
|
}
|
||
|
|
get _doesIntervalAllowAnother() {
|
||
|
|
return this._isIntervalIgnored || this._intervalCount < this._intervalCap;
|
||
|
|
}
|
||
|
|
get _doesConcurrentAllowAnother() {
|
||
|
|
return this._pendingCount < this._concurrency;
|
||
|
|
}
|
||
|
|
_next() {
|
||
|
|
this._pendingCount--;
|
||
|
|
this._tryToStartAnother();
|
||
|
|
this.emit('next');
|
||
|
|
}
|
||
|
|
_resolvePromises() {
|
||
|
|
this._resolveEmpty();
|
||
|
|
this._resolveEmpty = empty;
|
||
|
|
if (this._pendingCount === 0) {
|
||
|
|
this._resolveIdle();
|
||
|
|
this._resolveIdle = empty;
|
||
|
|
this.emit('idle');
|
||
|
|
}
|
||
|
|
}
|
||
|
|
_onResumeInterval() {
|
||
|
|
this._onInterval();
|
||
|
|
this._initializeIntervalIfNeeded();
|
||
|
|
this._timeoutId = undefined;
|
||
|
|
}
|
||
|
|
_isIntervalPaused() {
|
||
|
|
const now = Date.now();
|
||
|
|
if (this._intervalId === undefined) {
|
||
|
|
const delay = this._intervalEnd - now;
|
||
|
|
if (delay < 0) {
|
||
|
|
// Act as the interval was done
|
||
|
|
// We don't need to resume it here because it will be resumed on line 160
|
||
|
|
this._intervalCount = (this._carryoverConcurrencyCount) ? this._pendingCount : 0;
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
// Act as the interval is pending
|
||
|
|
if (this._timeoutId === undefined) {
|
||
|
|
this._timeoutId = setTimeout(() => {
|
||
|
|
this._onResumeInterval();
|
||
|
|
}, delay);
|
||
|
|
}
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
_tryToStartAnother() {
|
||
|
|
if (this._queue.size === 0) {
|
||
|
|
// We can clear the interval ("pause")
|
||
|
|
// Because we can redo it later ("resume")
|
||
|
|
if (this._intervalId) {
|
||
|
|
clearInterval(this._intervalId);
|
||
|
|
}
|
||
|
|
this._intervalId = undefined;
|
||
|
|
this._resolvePromises();
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
if (!this._isPaused) {
|
||
|
|
const canInitializeInterval = !this._isIntervalPaused();
|
||
|
|
if (this._doesIntervalAllowAnother && this._doesConcurrentAllowAnother) {
|
||
|
|
const job = this._queue.dequeue();
|
||
|
|
if (!job) {
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
this.emit('active');
|
||
|
|
job();
|
||
|
|
if (canInitializeInterval) {
|
||
|
|
this._initializeIntervalIfNeeded();
|
||
|
|
}
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return false;
|
||
|
|
}
|
||
|
|
_initializeIntervalIfNeeded() {
|
||
|
|
if (this._isIntervalIgnored || this._intervalId !== undefined) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
this._intervalId = setInterval(() => {
|
||
|
|
this._onInterval();
|
||
|
|
}, this._interval);
|
||
|
|
this._intervalEnd = Date.now() + this._interval;
|
||
|
|
}
|
||
|
|
_onInterval() {
|
||
|
|
if (this._intervalCount === 0 && this._pendingCount === 0 && this._intervalId) {
|
||
|
|
clearInterval(this._intervalId);
|
||
|
|
this._intervalId = undefined;
|
||
|
|
}
|
||
|
|
this._intervalCount = this._carryoverConcurrencyCount ? this._pendingCount : 0;
|
||
|
|
this._processQueue();
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Executes all queued functions until it reaches the limit.
|
||
|
|
*/
|
||
|
|
_processQueue() {
|
||
|
|
// eslint-disable-next-line no-empty
|
||
|
|
while (this._tryToStartAnother()) { }
|
||
|
|
}
|
||
|
|
get concurrency() {
|
||
|
|
return this._concurrency;
|
||
|
|
}
|
||
|
|
set concurrency(newConcurrency) {
|
||
|
|
if (!(typeof newConcurrency === 'number' && newConcurrency >= 1)) {
|
||
|
|
throw new TypeError(`Expected \`concurrency\` to be a number from 1 and up, got \`${newConcurrency}\` (${typeof newConcurrency})`);
|
||
|
|
}
|
||
|
|
this._concurrency = newConcurrency;
|
||
|
|
this._processQueue();
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Adds a sync or async task to the queue. Always returns a promise.
|
||
|
|
*/
|
||
|
|
async add(fn, options = {}) {
|
||
|
|
return new Promise((resolve, reject) => {
|
||
|
|
const run = async () => {
|
||
|
|
this._pendingCount++;
|
||
|
|
this._intervalCount++;
|
||
|
|
try {
|
||
|
|
const operation = (this._timeout === undefined && options.timeout === undefined) ? fn() : p_timeout_1.default(Promise.resolve(fn()), (options.timeout === undefined ? this._timeout : options.timeout), () => {
|
||
|
|
if (options.throwOnTimeout === undefined ? this._throwOnTimeout : options.throwOnTimeout) {
|
||
|
|
reject(timeoutError);
|
||
|
|
}
|
||
|
|
return undefined;
|
||
|
|
});
|
||
|
|
resolve(await operation);
|
||
|
|
}
|
||
|
|
catch (error) {
|
||
|
|
reject(error);
|
||
|
|
}
|
||
|
|
this._next();
|
||
|
|
};
|
||
|
|
this._queue.enqueue(run, options);
|
||
|
|
this._tryToStartAnother();
|
||
|
|
this.emit('add');
|
||
|
|
});
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Same as `.add()`, but accepts an array of sync or async functions.
|
||
|
|
|
||
|
|
@returns A promise that resolves when all functions are resolved.
|
||
|
|
*/
|
||
|
|
async addAll(functions, options) {
|
||
|
|
return Promise.all(functions.map(async (function_) => this.add(function_, options)));
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Start (or resume) executing enqueued tasks within concurrency limit. No need to call this if queue is not paused (via `options.autoStart = false` or by `.pause()` method.)
|
||
|
|
*/
|
||
|
|
start() {
|
||
|
|
if (!this._isPaused) {
|
||
|
|
return this;
|
||
|
|
}
|
||
|
|
this._isPaused = false;
|
||
|
|
this._processQueue();
|
||
|
|
return this;
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Put queue execution on hold.
|
||
|
|
*/
|
||
|
|
pause() {
|
||
|
|
this._isPaused = true;
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Clear the queue.
|
||
|
|
*/
|
||
|
|
clear() {
|
||
|
|
this._queue = new this._queueClass();
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Can be called multiple times. Useful if you for example add additional items at a later time.
|
||
|
|
|
||
|
|
@returns A promise that settles when the queue becomes empty.
|
||
|
|
*/
|
||
|
|
async onEmpty() {
|
||
|
|
// Instantly resolve if the queue is empty
|
||
|
|
if (this._queue.size === 0) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
return new Promise(resolve => {
|
||
|
|
const existingResolve = this._resolveEmpty;
|
||
|
|
this._resolveEmpty = () => {
|
||
|
|
existingResolve();
|
||
|
|
resolve();
|
||
|
|
};
|
||
|
|
});
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
The difference with `.onEmpty` is that `.onIdle` guarantees that all work from the queue has finished. `.onEmpty` merely signals that the queue is empty, but it could mean that some promises haven't completed yet.
|
||
|
|
|
||
|
|
@returns A promise that settles when the queue becomes empty, and all promises have completed; `queue.size === 0 && queue.pending === 0`.
|
||
|
|
*/
|
||
|
|
async onIdle() {
|
||
|
|
// Instantly resolve if none pending and if nothing else is queued
|
||
|
|
if (this._pendingCount === 0 && this._queue.size === 0) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
return new Promise(resolve => {
|
||
|
|
const existingResolve = this._resolveIdle;
|
||
|
|
this._resolveIdle = () => {
|
||
|
|
existingResolve();
|
||
|
|
resolve();
|
||
|
|
};
|
||
|
|
});
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Size of the queue.
|
||
|
|
*/
|
||
|
|
get size() {
|
||
|
|
return this._queue.size;
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Size of the queue, filtered by the given options.
|
||
|
|
|
||
|
|
For example, this can be used to find the number of items remaining in the queue with a specific priority level.
|
||
|
|
*/
|
||
|
|
sizeBy(options) {
|
||
|
|
// eslint-disable-next-line unicorn/no-fn-reference-in-iterator
|
||
|
|
return this._queue.filter(options).length;
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Number of pending promises.
|
||
|
|
*/
|
||
|
|
get pending() {
|
||
|
|
return this._pendingCount;
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Whether the queue is currently paused.
|
||
|
|
*/
|
||
|
|
get isPaused() {
|
||
|
|
return this._isPaused;
|
||
|
|
}
|
||
|
|
get timeout() {
|
||
|
|
return this._timeout;
|
||
|
|
}
|
||
|
|
/**
|
||
|
|
Set the timeout for future operations.
|
||
|
|
*/
|
||
|
|
set timeout(milliseconds) {
|
||
|
|
this._timeout = milliseconds;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
exports.default = PQueue;
|