Huge News!Announcing our $40M Series B led by Abstract Ventures.Learn More →
Socket
Sign inDemoInstall
Socket
Sign inDemoInstall

p-queue

Package Overview
Dependencies
Maintainers
2
Versions
44
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

p-queue - npm Package Compare versions

Comparing version 6.6.2 to 7.3.4

82

dist/index.d.ts

@@ -1,41 +0,24 @@

import EventEmitter = require('eventemitter3');
import { Queue, RunFunction } from './queue';
import PriorityQueue from './priority-queue';
import { QueueAddOptions, DefaultAddOptions, Options } from './options';
declare type Task<TaskResultType> = (() => PromiseLike<TaskResultType>) | (() => TaskResultType);
import EventEmitter from 'eventemitter3';
import { Queue, RunFunction } from './queue.js';
import PriorityQueue from './priority-queue.js';
import { QueueAddOptions, Options, TaskOptions } from './options.js';
type Task<TaskResultType> = ((options: TaskOptions) => PromiseLike<TaskResultType>) | ((options: TaskOptions) => TaskResultType);
/**
The error thrown by `queue.add()` when a job is aborted before it is run. See `signal`.
*/
export declare class AbortError extends Error {
}
type EventName = 'active' | 'idle' | 'empty' | 'add' | 'next' | 'completed' | 'error';
/**
Promise queue with concurrency control.
*/
export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsType> = PriorityQueue, EnqueueOptionsType extends QueueAddOptions = DefaultAddOptions> extends EventEmitter<'active' | 'idle' | 'add' | 'next'> {
private readonly _carryoverConcurrencyCount;
private readonly _isIntervalIgnored;
private _intervalCount;
private readonly _intervalCap;
private readonly _interval;
private _intervalEnd;
private _intervalId?;
private _timeoutId?;
private _queue;
private readonly _queueClass;
private _pendingCount;
private _concurrency;
private _isPaused;
private _resolveEmpty;
private _resolveIdle;
private _timeout?;
private readonly _throwOnTimeout;
constructor(options?: Options<QueueType, EnqueueOptionsType>);
private get _doesIntervalAllowAnother();
private get _doesConcurrentAllowAnother();
private _next;
private _resolvePromises;
private _onResumeInterval;
private _isIntervalPaused;
private _tryToStartAnother;
private _initializeIntervalIfNeeded;
private _onInterval;
export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsType> = PriorityQueue, EnqueueOptionsType extends QueueAddOptions = QueueAddOptions> extends EventEmitter<EventName> {
#private;
/**
Executes all queued functions until it reaches the limit.
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
Applies to each future operation.
*/
private _processQueue;
timeout?: number;
constructor(options?: Options<QueueType, EnqueueOptionsType>);
get concurrency(): number;

@@ -46,3 +29,6 @@ set concurrency(newConcurrency: number);

*/
add<TaskResultType>(fn: Task<TaskResultType>, options?: Partial<EnqueueOptionsType>): Promise<TaskResultType>;
add<TaskResultType>(function_: Task<TaskResultType>, options: {
throwOnTimeout: true;
} & Exclude<EnqueueOptionsType, 'throwOnTimeout'>): Promise<TaskResultType>;
add<TaskResultType>(function_: Task<TaskResultType>, options?: Partial<EnqueueOptionsType>): Promise<TaskResultType | void>;
/**

@@ -53,3 +39,6 @@ Same as `.add()`, but accepts an array of sync or async functions.

*/
addAll<TaskResultsType>(functions: ReadonlyArray<Task<TaskResultsType>>, options?: EnqueueOptionsType): Promise<TaskResultsType[]>;
addAll<TaskResultsType>(functions: ReadonlyArray<Task<TaskResultsType>>, options?: {
throwOnTimeout: true;
} & Partial<Exclude<EnqueueOptionsType, 'throwOnTimeout'>>): Promise<TaskResultsType[]>;
addAll<TaskResultsType>(functions: ReadonlyArray<Task<TaskResultsType>>, options?: Partial<EnqueueOptionsType>): Promise<Array<TaskResultsType | void>>;
/**

@@ -74,2 +63,10 @@ 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.)

/**
@returns A promise that settles when the queue size is less than the given limit: `queue.size < limit`.
If you want to avoid having the queue grow beyond a certain size you can `await queue.onSizeLessThan()` before adding a new item.
Note that this only limits the number of items waiting to start. There could still be up to `concurrency` jobs already running that this call does not include in its calculation.
*/
onSizeLessThan(limit: number): Promise<void>;
/**
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.

@@ -81,3 +78,3 @@

/**
Size of the queue.
Size of the queue, the number of queued items waiting to run.
*/

@@ -92,3 +89,3 @@ get size(): number;

/**
Number of pending promises.
Number of running items (no longer in the queue).
*/

@@ -100,8 +97,3 @@ get pending(): number;

get isPaused(): boolean;
get timeout(): number | undefined;
/**
Set the timeout for future operations.
*/
set timeout(milliseconds: number | undefined);
}
export { Queue, QueueAddOptions, DefaultAddOptions, Options };
export { Queue, QueueAddOptions, QueueAddOptions as DefaultAddOptions, Options };

421

dist/index.js

@@ -1,23 +0,66 @@

"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();
var __classPrivateFieldSet = (this && this.__classPrivateFieldSet) || function (receiver, state, value, kind, f) {
if (kind === "m") throw new TypeError("Private method is not writable");
if (kind === "a" && !f) throw new TypeError("Private accessor was defined without a setter");
if (typeof state === "function" ? receiver !== state || !f : !state.has(receiver)) throw new TypeError("Cannot write private member to an object whose class did not declare it");
return (kind === "a" ? f.call(receiver, value) : f ? f.value = value : state.set(receiver, value)), value;
};
var __classPrivateFieldGet = (this && this.__classPrivateFieldGet) || function (receiver, state, kind, f) {
if (kind === "a" && !f) throw new TypeError("Private accessor was defined without a getter");
if (typeof state === "function" ? receiver !== state || !f : !state.has(receiver)) throw new TypeError("Cannot read private member from an object whose class did not declare it");
return kind === "m" ? f : kind === "a" ? f.call(receiver) : f ? f.value : state.get(receiver);
};
var _PQueue_instances, _PQueue_carryoverConcurrencyCount, _PQueue_isIntervalIgnored, _PQueue_intervalCount, _PQueue_intervalCap, _PQueue_interval, _PQueue_intervalEnd, _PQueue_intervalId, _PQueue_timeoutId, _PQueue_queue, _PQueue_queueClass, _PQueue_pending, _PQueue_concurrency, _PQueue_isPaused, _PQueue_throwOnTimeout, _PQueue_doesIntervalAllowAnother_get, _PQueue_doesConcurrentAllowAnother_get, _PQueue_next, _PQueue_onResumeInterval, _PQueue_isIntervalPaused_get, _PQueue_tryToStartAnother, _PQueue_initializeIntervalIfNeeded, _PQueue_onInterval, _PQueue_processQueue, _PQueue_throwOnAbort, _PQueue_onEvent;
import EventEmitter from 'eventemitter3';
import pTimeout, { TimeoutError } from 'p-timeout';
import PriorityQueue from './priority-queue.js';
/**
The error thrown by `queue.add()` when a job is aborted before it is run. See `signal`.
*/
export class AbortError extends Error {
}
/**
Promise queue with concurrency control.
*/
class PQueue extends EventEmitter {
export default class PQueue extends EventEmitter {
// TODO: The `throwOnTimeout` option should affect the return types of `add()` and `addAll()`
constructor(options) {
var _a, _b, _c, _d;
super();
this._intervalCount = 0;
this._intervalEnd = 0;
this._pendingCount = 0;
this._resolveEmpty = empty;
this._resolveIdle = empty;
_PQueue_instances.add(this);
_PQueue_carryoverConcurrencyCount.set(this, void 0);
_PQueue_isIntervalIgnored.set(this, void 0);
_PQueue_intervalCount.set(this, 0);
_PQueue_intervalCap.set(this, void 0);
_PQueue_interval.set(this, void 0);
_PQueue_intervalEnd.set(this, 0);
_PQueue_intervalId.set(this, void 0);
_PQueue_timeoutId.set(this, void 0);
_PQueue_queue.set(this, void 0);
_PQueue_queueClass.set(this, void 0);
_PQueue_pending.set(this, 0);
// The `!` is needed because of https://github.com/microsoft/TypeScript/issues/32194
_PQueue_concurrency.set(this, void 0);
_PQueue_isPaused.set(this, void 0);
_PQueue_throwOnTimeout.set(this, void 0);
/**
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
Applies to each future operation.
*/
Object.defineProperty(this, "timeout", {
enumerable: true,
configurable: true,
writable: true,
value: void 0
});
// 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);
options = {
carryoverConcurrencyCount: false,
intervalCap: Number.POSITIVE_INFINITY,
interval: 0,
concurrency: Number.POSITIVE_INFINITY,
autoStart: true,
queueClass: PriorityQueue,
...options,
};
if (!(typeof options.intervalCap === 'number' && options.intervalCap >= 1)) {

@@ -29,113 +72,15 @@ 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})`);

}
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;
__classPrivateFieldSet(this, _PQueue_carryoverConcurrencyCount, options.carryoverConcurrencyCount, "f");
__classPrivateFieldSet(this, _PQueue_isIntervalIgnored, options.intervalCap === Number.POSITIVE_INFINITY || options.interval === 0, "f");
__classPrivateFieldSet(this, _PQueue_intervalCap, options.intervalCap, "f");
__classPrivateFieldSet(this, _PQueue_interval, options.interval, "f");
__classPrivateFieldSet(this, _PQueue_queue, new options.queueClass(), "f");
__classPrivateFieldSet(this, _PQueue_queueClass, options.queueClass, "f");
this.concurrency = options.concurrency;
this._timeout = options.timeout;
this._throwOnTimeout = options.throwOnTimeout === true;
this._isPaused = options.autoStart === false;
this.timeout = options.timeout;
__classPrivateFieldSet(this, _PQueue_throwOnTimeout, options.throwOnTimeout === true, "f");
__classPrivateFieldSet(this, _PQueue_isPaused, options.autoStart === false, "f");
}
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;
return __classPrivateFieldGet(this, _PQueue_concurrency, "f");
}

@@ -146,37 +91,50 @@ set concurrency(newConcurrency) {

}
this._concurrency = newConcurrency;
this._processQueue();
__classPrivateFieldSet(this, _PQueue_concurrency, newConcurrency, "f");
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_processQueue).call(this);
}
/**
Adds a sync or async task to the queue. Always returns a promise.
*/
async add(fn, options = {}) {
async add(function_, options = {}) {
options = {
timeout: this.timeout,
throwOnTimeout: __classPrivateFieldGet(this, _PQueue_throwOnTimeout, "f"),
...options,
};
return new Promise((resolve, reject) => {
const run = async () => {
this._pendingCount++;
this._intervalCount++;
__classPrivateFieldGet(this, _PQueue_queue, "f").enqueue(async () => {
var _a;
var _b, _c;
__classPrivateFieldSet(this, _PQueue_pending, (_b = __classPrivateFieldGet(this, _PQueue_pending, "f"), _b++, _b), "f");
__classPrivateFieldSet(this, _PQueue_intervalCount, (_c = __classPrivateFieldGet(this, _PQueue_intervalCount, "f"), _c++, _c), "f");
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);
// TODO: Use options.signal?.throwIfAborted() when targeting Node.js 18
if ((_a = options.signal) === null || _a === void 0 ? void 0 : _a.aborted) {
// TODO: Use ABORT_ERR code when targeting Node.js 16 (https://nodejs.org/docs/latest-v16.x/api/errors.html#abort_err)
throw new AbortError('The task was aborted.');
}
let operation = function_({ signal: options.signal });
if (options.timeout) {
operation = pTimeout(Promise.resolve(operation), options.timeout);
}
if (options.signal) {
operation = Promise.race([operation, __classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_throwOnAbort).call(this, options.signal)]);
}
const result = await operation;
resolve(result);
this.emit('completed', result);
}
catch (error) {
if (error instanceof TimeoutError && !options.throwOnTimeout) {
resolve();
return;
}
reject(error);
this.emit('error', error);
}
this._next();
};
this._queue.enqueue(run, options);
this._tryToStartAnother();
finally {
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_next).call(this);
}
}, options);
this.emit('add');
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_tryToStartAnother).call(this);
});
}
/**
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) {

@@ -189,7 +147,7 @@ return Promise.all(functions.map(async (function_) => this.add(function_, options)));

start() {
if (!this._isPaused) {
if (!__classPrivateFieldGet(this, _PQueue_isPaused, "f")) {
return this;
}
this._isPaused = false;
this._processQueue();
__classPrivateFieldSet(this, _PQueue_isPaused, false, "f");
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_processQueue).call(this);
return this;

@@ -201,3 +159,3 @@ }

pause() {
this._isPaused = true;
__classPrivateFieldSet(this, _PQueue_isPaused, true, "f");
}

@@ -208,3 +166,3 @@ /**

clear() {
this._queue = new this._queueClass();
__classPrivateFieldSet(this, _PQueue_queue, new (__classPrivateFieldGet(this, _PQueue_queueClass, "f"))(), "f");
}

@@ -218,14 +176,22 @@ /**

// Instantly resolve if the queue is empty
if (this._queue.size === 0) {
if (__classPrivateFieldGet(this, _PQueue_queue, "f").size === 0) {
return;
}
return new Promise(resolve => {
const existingResolve = this._resolveEmpty;
this._resolveEmpty = () => {
existingResolve();
resolve();
};
});
await __classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_onEvent).call(this, 'empty');
}
/**
@returns A promise that settles when the queue size is less than the given limit: `queue.size < limit`.
If you want to avoid having the queue grow beyond a certain size you can `await queue.onSizeLessThan()` before adding a new item.
Note that this only limits the number of items waiting to start. There could still be up to `concurrency` jobs already running that this call does not include in its calculation.
*/
async onSizeLessThan(limit) {
// Instantly resolve if the queue is empty.
if (__classPrivateFieldGet(this, _PQueue_queue, "f").size < limit) {
return;
}
await __classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_onEvent).call(this, 'next', () => __classPrivateFieldGet(this, _PQueue_queue, "f").size < limit);
}
/**
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.

@@ -237,18 +203,12 @@

// Instantly resolve if none pending and if nothing else is queued
if (this._pendingCount === 0 && this._queue.size === 0) {
if (__classPrivateFieldGet(this, _PQueue_pending, "f") === 0 && __classPrivateFieldGet(this, _PQueue_queue, "f").size === 0) {
return;
}
return new Promise(resolve => {
const existingResolve = this._resolveIdle;
this._resolveIdle = () => {
existingResolve();
resolve();
};
});
await __classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_onEvent).call(this, 'idle');
}
/**
Size of the queue.
Size of the queue, the number of queued items waiting to run.
*/
get size() {
return this._queue.size;
return __classPrivateFieldGet(this, _PQueue_queue, "f").size;
}

@@ -261,10 +221,10 @@ /**

sizeBy(options) {
// eslint-disable-next-line unicorn/no-fn-reference-in-iterator
return this._queue.filter(options).length;
// eslint-disable-next-line unicorn/no-array-callback-reference
return __classPrivateFieldGet(this, _PQueue_queue, "f").filter(options).length;
}
/**
Number of pending promises.
Number of running items (no longer in the queue).
*/
get pending() {
return this._pendingCount;
return __classPrivateFieldGet(this, _PQueue_pending, "f");
}

@@ -275,14 +235,105 @@ /**

get isPaused() {
return this._isPaused;
return __classPrivateFieldGet(this, _PQueue_isPaused, "f");
}
get timeout() {
return this._timeout;
}
_PQueue_carryoverConcurrencyCount = new WeakMap(), _PQueue_isIntervalIgnored = new WeakMap(), _PQueue_intervalCount = new WeakMap(), _PQueue_intervalCap = new WeakMap(), _PQueue_interval = new WeakMap(), _PQueue_intervalEnd = new WeakMap(), _PQueue_intervalId = new WeakMap(), _PQueue_timeoutId = new WeakMap(), _PQueue_queue = new WeakMap(), _PQueue_queueClass = new WeakMap(), _PQueue_pending = new WeakMap(), _PQueue_concurrency = new WeakMap(), _PQueue_isPaused = new WeakMap(), _PQueue_throwOnTimeout = new WeakMap(), _PQueue_instances = new WeakSet(), _PQueue_doesIntervalAllowAnother_get = function _PQueue_doesIntervalAllowAnother_get() {
return __classPrivateFieldGet(this, _PQueue_isIntervalIgnored, "f") || __classPrivateFieldGet(this, _PQueue_intervalCount, "f") < __classPrivateFieldGet(this, _PQueue_intervalCap, "f");
}, _PQueue_doesConcurrentAllowAnother_get = function _PQueue_doesConcurrentAllowAnother_get() {
return __classPrivateFieldGet(this, _PQueue_pending, "f") < __classPrivateFieldGet(this, _PQueue_concurrency, "f");
}, _PQueue_next = function _PQueue_next() {
var _a;
__classPrivateFieldSet(this, _PQueue_pending, (_a = __classPrivateFieldGet(this, _PQueue_pending, "f"), _a--, _a), "f");
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_tryToStartAnother).call(this);
this.emit('next');
}, _PQueue_onResumeInterval = function _PQueue_onResumeInterval() {
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_onInterval).call(this);
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_initializeIntervalIfNeeded).call(this);
__classPrivateFieldSet(this, _PQueue_timeoutId, undefined, "f");
}, _PQueue_isIntervalPaused_get = function _PQueue_isIntervalPaused_get() {
const now = Date.now();
if (__classPrivateFieldGet(this, _PQueue_intervalId, "f") === undefined) {
const delay = __classPrivateFieldGet(this, _PQueue_intervalEnd, "f") - 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
__classPrivateFieldSet(this, _PQueue_intervalCount, (__classPrivateFieldGet(this, _PQueue_carryoverConcurrencyCount, "f")) ? __classPrivateFieldGet(this, _PQueue_pending, "f") : 0, "f");
}
else {
// Act as the interval is pending
if (__classPrivateFieldGet(this, _PQueue_timeoutId, "f") === undefined) {
__classPrivateFieldSet(this, _PQueue_timeoutId, setTimeout(() => {
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_onResumeInterval).call(this);
}, delay), "f");
}
return true;
}
}
/**
Set the timeout for future operations.
*/
set timeout(milliseconds) {
this._timeout = milliseconds;
return false;
}, _PQueue_tryToStartAnother = function _PQueue_tryToStartAnother() {
if (__classPrivateFieldGet(this, _PQueue_queue, "f").size === 0) {
// We can clear the interval ("pause")
// Because we can redo it later ("resume")
if (__classPrivateFieldGet(this, _PQueue_intervalId, "f")) {
clearInterval(__classPrivateFieldGet(this, _PQueue_intervalId, "f"));
}
__classPrivateFieldSet(this, _PQueue_intervalId, undefined, "f");
this.emit('empty');
if (__classPrivateFieldGet(this, _PQueue_pending, "f") === 0) {
this.emit('idle');
}
return false;
}
}
exports.default = PQueue;
if (!__classPrivateFieldGet(this, _PQueue_isPaused, "f")) {
const canInitializeInterval = !__classPrivateFieldGet(this, _PQueue_instances, "a", _PQueue_isIntervalPaused_get);
if (__classPrivateFieldGet(this, _PQueue_instances, "a", _PQueue_doesIntervalAllowAnother_get) && __classPrivateFieldGet(this, _PQueue_instances, "a", _PQueue_doesConcurrentAllowAnother_get)) {
const job = __classPrivateFieldGet(this, _PQueue_queue, "f").dequeue();
if (!job) {
return false;
}
this.emit('active');
job();
if (canInitializeInterval) {
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_initializeIntervalIfNeeded).call(this);
}
return true;
}
}
return false;
}, _PQueue_initializeIntervalIfNeeded = function _PQueue_initializeIntervalIfNeeded() {
if (__classPrivateFieldGet(this, _PQueue_isIntervalIgnored, "f") || __classPrivateFieldGet(this, _PQueue_intervalId, "f") !== undefined) {
return;
}
__classPrivateFieldSet(this, _PQueue_intervalId, setInterval(() => {
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_onInterval).call(this);
}, __classPrivateFieldGet(this, _PQueue_interval, "f")), "f");
__classPrivateFieldSet(this, _PQueue_intervalEnd, Date.now() + __classPrivateFieldGet(this, _PQueue_interval, "f"), "f");
}, _PQueue_onInterval = function _PQueue_onInterval() {
if (__classPrivateFieldGet(this, _PQueue_intervalCount, "f") === 0 && __classPrivateFieldGet(this, _PQueue_pending, "f") === 0 && __classPrivateFieldGet(this, _PQueue_intervalId, "f")) {
clearInterval(__classPrivateFieldGet(this, _PQueue_intervalId, "f"));
__classPrivateFieldSet(this, _PQueue_intervalId, undefined, "f");
}
__classPrivateFieldSet(this, _PQueue_intervalCount, __classPrivateFieldGet(this, _PQueue_carryoverConcurrencyCount, "f") ? __classPrivateFieldGet(this, _PQueue_pending, "f") : 0, "f");
__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_processQueue).call(this);
}, _PQueue_processQueue = function _PQueue_processQueue() {
// eslint-disable-next-line no-empty
while (__classPrivateFieldGet(this, _PQueue_instances, "m", _PQueue_tryToStartAnother).call(this)) { }
}, _PQueue_throwOnAbort = async function _PQueue_throwOnAbort(signal) {
return new Promise((_resolve, reject) => {
signal.addEventListener('abort', () => {
// TODO: Reject with signal.throwIfAborted() when targeting Node.js 18
// TODO: Use ABORT_ERR code when targeting Node.js 16 (https://nodejs.org/docs/latest-v16.x/api/errors.html#abort_err)
reject(new AbortError('The task was aborted.'));
}, { once: true });
});
}, _PQueue_onEvent = async function _PQueue_onEvent(event, filter) {
return new Promise(resolve => {
const listener = () => {
if (filter && !filter()) {
return;
}
this.off(event, listener);
resolve();
};
this.on(event, listener);
});
};

7

dist/lower-bound.js

@@ -1,10 +0,8 @@

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
// Port of lower_bound from https://en.cppreference.com/w/cpp/algorithm/lower_bound
// Used to compute insertion index to keep queue sorted after insertion
function lowerBound(array, value, comparator) {
export default function lowerBound(array, value, comparator) {
let first = 0;
let count = array.length;
while (count > 0) {
const step = (count / 2) | 0;
const step = Math.trunc(count / 2);
let it = first + step;

@@ -21,2 +19,1 @@ if (comparator(array[it], value) <= 0) {

}
exports.default = lowerBound;

68

dist/options.d.ts

@@ -1,6 +0,15 @@

import { Queue, RunFunction } from './queue';
export interface QueueAddOptions {
readonly [key: string]: unknown;
import { Queue, RunFunction } from './queue.js';
interface TimeoutOptions {
/**
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
*/
timeout?: number;
/**
Whether or not a timeout is considered an exception.
@default false
*/
throwOnTimeout?: boolean;
}
export interface Options<QueueType extends Queue<RunFunction, QueueOptions>, QueueOptions extends QueueAddOptions> {
export interface Options<QueueType extends Queue<RunFunction, QueueOptions>, QueueOptions extends QueueAddOptions> extends TimeoutOptions {
/**

@@ -46,14 +55,4 @@ Concurrency limit.

readonly carryoverConcurrencyCount?: boolean;
/**
Per-operation timeout in milliseconds. Operations fulfill once `timeout` elapses if they haven't already.
*/
timeout?: number;
/**
Whether or not a timeout is considered an exception.
@default false
*/
throwOnTimeout?: boolean;
}
export interface DefaultAddOptions extends QueueAddOptions {
export interface QueueAddOptions extends TaskOptions, TimeoutOptions {
/**

@@ -66,1 +65,40 @@ Priority of operation. Operations with greater priority will be scheduled first.

}
export interface TaskOptions {
/**
[`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal) for cancellation of the operation. When aborted, it will be removed from the queue and the `queue.add()` call will reject with an `AbortError`. If the operation is already running, the signal will need to be handled by the operation itself.
@example
```
import PQueue, {AbortError} from 'p-queue';
import got, {CancelError} from 'got';
const queue = new PQueue();
const controller = new AbortController();
try {
await queue.add(({signal}) => {
const request = got('https://sindresorhus.com');
signal.addEventListener('abort', () => {
request.cancel();
});
try {
return await request;
} catch (error) {
if (!(error instanceof CancelError)) {
throw error;
}
}
}, {signal: controller.signal});
} catch (error) {
if (!(error instanceof AbortError)) {
throw error;
}
}
```
*/
readonly signal?: AbortSignal;
}
export {};

3

dist/options.js

@@ -1,2 +0,1 @@

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

6

dist/priority-queue.d.ts

@@ -1,3 +0,3 @@

import { Queue, RunFunction } from './queue';
import { QueueAddOptions } from './options';
import { Queue, RunFunction } from './queue.js';
import { QueueAddOptions } from './options.js';
export interface PriorityQueueOptions extends QueueAddOptions {

@@ -7,3 +7,3 @@ priority?: number;

export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOptions> {
private readonly _queue;
#private;
enqueue(run: RunFunction, options?: Partial<PriorityQueueOptions>): void;

@@ -10,0 +10,0 @@ dequeue(): RunFunction | undefined;

37

dist/priority-queue.js

@@ -1,32 +0,39 @@

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const lower_bound_1 = require("./lower-bound");
class PriorityQueue {
var __classPrivateFieldGet = (this && this.__classPrivateFieldGet) || function (receiver, state, kind, f) {
if (kind === "a" && !f) throw new TypeError("Private accessor was defined without a getter");
if (typeof state === "function" ? receiver !== state || !f : !state.has(receiver)) throw new TypeError("Cannot read private member from an object whose class did not declare it");
return kind === "m" ? f : kind === "a" ? f.call(receiver) : f ? f.value : state.get(receiver);
};
var _PriorityQueue_queue;
import lowerBound from './lower-bound.js';
export default class PriorityQueue {
constructor() {
this._queue = [];
_PriorityQueue_queue.set(this, []);
}
enqueue(run, options) {
options = Object.assign({ priority: 0 }, options);
options = {
priority: 0,
...options,
};
const element = {
priority: options.priority,
run
run,
};
if (this.size && this._queue[this.size - 1].priority >= options.priority) {
this._queue.push(element);
if (this.size && __classPrivateFieldGet(this, _PriorityQueue_queue, "f")[this.size - 1].priority >= options.priority) {
__classPrivateFieldGet(this, _PriorityQueue_queue, "f").push(element);
return;
}
const index = lower_bound_1.default(this._queue, element, (a, b) => b.priority - a.priority);
this._queue.splice(index, 0, element);
const index = lowerBound(__classPrivateFieldGet(this, _PriorityQueue_queue, "f"), element, (a, b) => b.priority - a.priority);
__classPrivateFieldGet(this, _PriorityQueue_queue, "f").splice(index, 0, element);
}
dequeue() {
const item = this._queue.shift();
const item = __classPrivateFieldGet(this, _PriorityQueue_queue, "f").shift();
return item === null || item === void 0 ? void 0 : item.run;
}
filter(options) {
return this._queue.filter((element) => element.priority === options.priority).map((element) => element.run);
return __classPrivateFieldGet(this, _PriorityQueue_queue, "f").filter((element) => element.priority === options.priority).map((element) => element.run);
}
get size() {
return this._queue.length;
return __classPrivateFieldGet(this, _PriorityQueue_queue, "f").length;
}
}
exports.default = PriorityQueue;
_PriorityQueue_queue = new WeakMap();

2

dist/queue.d.ts

@@ -1,2 +0,2 @@

export declare type RunFunction = () => Promise<unknown>;
export type RunFunction = () => Promise<unknown>;
export interface Queue<Element, Options> {

@@ -3,0 +3,0 @@ size: number;

3

dist/queue.js

@@ -1,2 +0,1 @@

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

68

package.json
{
"name": "p-queue",
"version": "6.6.2",
"version": "7.3.4",
"description": "Promise queue with concurrency control",

@@ -8,11 +8,12 @@ "license": "MIT",

"funding": "https://github.com/sponsors/sindresorhus",
"main": "dist/index.js",
"type": "module",
"exports": "./dist/index.js",
"engines": {
"node": ">=8"
"node": ">=12"
},
"scripts": {
"build": "del dist && tsc",
"test": "xo && npm run build && nyc ava",
"bench": "ts-node bench.ts",
"prepublishOnly": "npm run build"
"build": "del-cli dist && tsc",
"test": "xo && ava && del-cli dist && tsc && tsd",
"bench": "node --loader=ts-node/esm bench.ts",
"prepublishOnly": "del-cli dist && tsc"
},

@@ -46,33 +47,31 @@ "files": [

"dependencies": {
"eventemitter3": "^4.0.4",
"p-timeout": "^3.2.0"
"eventemitter3": "^4.0.7",
"p-timeout": "^5.0.2"
},
"devDependencies": {
"@sindresorhus/tsconfig": "^0.7.0",
"@types/benchmark": "^1.0.33",
"@types/node": "^14.6.0",
"ava": "^2.0.0",
"@sindresorhus/tsconfig": "^2.0.0",
"@types/benchmark": "^2.1.1",
"@types/node": "^17.0.13",
"ava": "^5.1.1",
"benchmark": "^2.1.4",
"codecov": "^3.7.2",
"del-cli": "^3.0.1",
"delay": "^4.4.0",
"in-range": "^2.0.0",
"nyc": "^15.1.0",
"random-int": "^2.0.1",
"time-span": "^4.0.0",
"ts-node": "^9.0.0",
"typescript": "^4.0.2",
"xo": "^0.33.0"
"del-cli": "^5.0.0",
"delay": "^5.0.0",
"in-range": "^3.0.0",
"p-defer": "^4.0.0",
"random-int": "^3.0.0",
"time-span": "^5.0.0",
"ts-node": "^10.9.1",
"tsd": "^0.25.0",
"typescript": "^4.8.4",
"xo": "^0.44.0"
},
"ava": {
"babel": false,
"compileEnhancements": false,
"extensions": [
"ts"
],
"require": [
"ts-node/register"
],
"files": [
"test/**"
],
"extensions": {
"ts": "module"
},
"nodeArguments": [
"--loader=ts-node/esm"
]

@@ -83,13 +82,6 @@ },

"@typescript-eslint/member-ordering": "off",
"node/no-unsupported-features/es-syntax": "off",
"@typescript-eslint/no-floating-promises": "off",
"import/no-named-default": "off",
"@typescript-eslint/no-invalid-void-type": "off"
}
},
"nyc": {
"extension": [
".ts"
]
}
}

152

readme.md

@@ -1,2 +0,2 @@

# p-queue [![Build Status](https://travis-ci.com/sindresorhus/p-queue.svg?branch=master)](https://travis-ci.com/github/sindresorhus/p-queue) [![codecov](https://codecov.io/gh/sindresorhus/p-queue/branch/master/graph/badge.svg)](https://codecov.io/gh/sindresorhus/p-queue)
# p-queue

@@ -7,8 +7,14 @@ > Promise queue with concurrency control

For servers, you probably want a Redis-backed [job queue](https://github.com/sindresorhus/awesome-nodejs#job-queues) instead.
Note that the project is feature complete. We are happy to review pull requests, but we don't plan any further development. We are also not answering email support questions.
## Install
```sh
npm install p-queue
```
$ npm install p-queue
```
**Warning:** This package is native [ESM](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Guide/Modules) and no longer provides a CommonJS export. If your project uses CommonJS, you'll have to [convert to ESM](https://gist.github.com/sindresorhus/a39789f98801d908bbc7ff3ecc99d99c) or use the [dynamic `import()`](https://v8.dev/features/dynamic-import) function. Please don't open issues for questions regarding CommonJS / ESM. You can also use [version 6](https://github.com/sindresorhus/p-queue/tree/v6.6.2) instead which is pretty stable. We will backport security fixes to v6 for the foreseeable future.
## Usage

@@ -19,4 +25,4 @@

```js
const {default: PQueue} = require('p-queue');
const got = require('got');
import PQueue from 'p-queue';
import got from 'got';

@@ -117,2 +123,4 @@ const queue = new PQueue({concurrency: 1});

Note: If your items can potentially throw an exception, you must handle those errors from the returned Promise or they may be reported as an unhandled Promise rejection and potentially cause your process to exit immediately.
##### fn

@@ -122,3 +130,3 @@

Promise-returning/async function.
Promise-returning/async function. When executed, it will receive `{signal}` as the first argument.

@@ -136,2 +144,39 @@ #### options

##### signal
*Requires Node.js 16 or later.*
[`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal) for cancellation of the operation. When aborted, it will be removed from the queue and the `queue.add()` call will reject with an `AbortError`. If the operation is already running, the signal will need to be handled by the operation itself.
```js
import PQueue, {AbortError} from 'p-queue';
import got, {CancelError} from 'got';
const queue = new PQueue();
const controller = new AbortController();
try {
await queue.add(({signal}) => {
const request = got('https://sindresorhus.com');
signal.addEventListener('abort', () => {
request.cancel();
});
try {
return await request;
} catch (error) {
if (!(error instanceof CancelError)) {
throw error;
}
}
}, {signal: controller.signal});
} catch (error) {
if (!(error instanceof AbortError)) {
throw error;
}
}
```
#### .addAll(fns, options?)

@@ -163,2 +208,10 @@

#### .onSizeLessThan(limit)
Returns a promise that settles when the queue size is less than the given limit: `queue.size < limit`.
If you want to avoid having the queue grow beyond a certain size you can `await queue.onSizeLessThan()` before adding a new item.
Note that this only limits the number of items waiting to start. There could still be up to `concurrency` jobs already running that this call does not include in its calculation.
#### .clear()

@@ -170,3 +223,3 @@

Size of the queue.
Size of the queue, the number of queued items waiting to run.

@@ -180,2 +233,4 @@ #### .sizeBy(options)

```js
import PQueue from 'p-queue';
const queue = new PQueue();

@@ -196,3 +251,3 @@

Number of pending promises.
Number of running items (no longer in the queue).

@@ -214,4 +269,4 @@ #### [.timeout](#timeout)

```js
const delay = require('delay');
const {default: PQueue} = require('p-queue');
import delay from 'delay';
import PQueue from 'p-queue';

@@ -231,2 +286,43 @@ const queue = new PQueue({concurrency: 2});

```
#### completed
Emitted when an item completes without error.
```js
import delay from 'delay';
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
queue.on('completed', result => {
console.log(result);
});
queue.add(() => Promise.resolve('hello, world!'));
```
#### error
Emitted if an item throws an error.
```js
import delay from 'delay';
import PQueue from 'p-queue';
const queue = new PQueue({concurrency: 2});
queue.on('error', error => {
console.error(error);
});
queue.add(() => Promise.reject(new Error('error')));
```
#### empty
Emitted every time the queue becomes empty.
Useful if you for example add additional items at a later time.
#### idle

@@ -236,5 +332,7 @@

The difference with `empty` is that `idle` guarantees that all work from the queue has finished. `empty` merely signals that the queue is empty, but it could mean that some promises haven't completed yet.
```js
const delay = require('delay');
const {default: PQueue} = require('p-queue');
import delay from 'delay';
import PQueue from 'p-queue';

@@ -266,7 +364,7 @@ const queue = new PQueue();

Emitted every time a task is completed and the number of pending or queued tasks is decreased.
Emitted every time a task is completed and the number of pending or queued tasks is decreased. This is emitted regardless of whether the task completed normally or with an error.
```js
const delay = require('delay');
const {default: PQueue} = require('p-queue');
import delay from 'delay';
import PQueue from 'p-queue';

@@ -278,2 +376,3 @@ const queue = new PQueue();

});
queue.on('next', () => {

@@ -296,2 +395,6 @@ console.log(`Task is completed. Size: ${queue.size} Pending: ${queue.pending}`);

### AbortError
The error thrown by `queue.add()` when a job is aborted before it is run. See [`signal`](#signal).
## Advanced example

@@ -302,4 +405,4 @@

```js
const delay = require('delay');
const {default: PQueue} = require('p-queue');
import delay from 'delay';
import PQueue from 'p-queue';

@@ -373,2 +476,4 @@ const queue = new PQueue({concurrency: 1});

```js
import PQueue from 'p-queue';
class QueueClass {

@@ -395,2 +500,4 @@ constructor() {

}
const queue = new PQueue({queueClass: QueueClass});
```

@@ -400,2 +507,13 @@

## FAQ
#### How do the `concurrency` and `intervalCap` options affect each other?
They are just different constraints. The `concurrency` option limits how many things run at the same time. The `intervalCap` option limits how many things run in total during the interval (over time).
## Maintainers
- [Sindre Sorhus](https://github.com/sindresorhus)
- [Richie Bendall](https://github.com/Richienb)
## Related

@@ -402,0 +520,0 @@

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

About

Packages

npm

Go

Maven

PyPI

Rubygems

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc