timequeue - npm Package Compare versions

+81

dist/index.d.ts

		/// <reference types="node" />
		declare const EventEmitter: any;
		declare namespace TimeQueue {
		type Worker = (...args: any[]) => void \| Promise<any>;
		interface Options {
		concurrency?: number;
		every?: number;
		maxQueued?: number;
		timeout?: number;
		}
		interface TaskError extends Error {
		args: any[];
		}
		}
		declare class TimeQueue extends EventEmitter {
		worker: TimeQueue.Worker;
		concurrency: number;
		every: number;
		maxQueued: number;
		timeout: number;
		private _workerAsync;
		private _queue;
		private _timers;
		active: number;
		intransit: number;
		queued: number;
		finished: number;
		static TaskError: {
		new (message?: string): {
		args: any[];
		name: string;
		message: string;
		stack?: string;
		};
		captureStackTrace(targetObject: Object, constructorOpt?: Function): void;
		prepareStackTrace?: (err: Error, stackTraces: NodeJS.CallSite[]) => any;
		stackTraceLimit: number;
		};
		/**
		* @constructor
		* @extends {EventEmitter}
		* @param {Function(..., Function(!Error, ...)} worker
		* @param {Object?} options
		* @param {number?} options.concurrency
		* @param {number?} options.every
		* @param {number?} options.maxQueued
		* @param {number?} options.timeout
		*/
		constructor(worker: TimeQueue.Worker, options?: TimeQueue.Options);
		/**
		* Pushes a task onto the queue.
		*
		* @param {Object} ...args
		* @param {Function(!Error, ...)} callback
		* @return {Promise?}
		*/
		push(...args: any[]): Promise<unknown>;
		/**
		* Returns true if queue is full.
		*
		* @return {boolean}
		*/
		isFull(): boolean;
		/**
		* Starts a task
		*
		* @param {Array.<Object>} args
		*/
		_process(args: any[]): Promise<void>;
		/**
		* Called when a task finishes. Looks at the queue and processes the next
		* waiting task.
		*/
		_next(): void;
		/**
		* Empties the queue and kills the timers.
		* Active tasks will still be completed.
		*/
		die(): void;
		}
		export = TimeQueue;

+196

dist/index.js

		"use strict";
		var __awaiter = (this && this.__awaiter) \|\| function (thisArg, _arguments, P, generator) {
		function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
		return new (P \|\| (P = Promise))(function (resolve, reject) {
		function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
		function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
		function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
		step((generator = generator.apply(thisArg, _arguments \|\| [])).next());
		});
		};
		const EventEmitter = require('events').EventEmitter;
		class TimeQueue extends EventEmitter {
		/**
		* @constructor
		* @extends {EventEmitter}
		* @param {Function(..., Function(!Error, ...)} worker
		* @param {Object?} options
		* @param {number?} options.concurrency
		* @param {number?} options.every
		* @param {number?} options.maxQueued
		* @param {number?} options.timeout
		*/
		constructor(worker, options = {}) {
		super();
		this.worker = worker;
		this._workerAsync = worker.constructor.name == 'AsyncFunction';
		this.concurrency = options.concurrency \|\| 1;
		this.every = options.every \|\| 0;
		this.maxQueued = options.maxQueued \|\| Infinity;
		this.timeout = options.timeout \|\| 0;
		this._queue = [];
		this._timers = [];
		this.active = 0;
		this.intransit = 0;
		this.queued = 0;
		this.finished = 0;
		}
		/**
		* Pushes a task onto the queue.
		*
		* @param {Object} ...args
		* @param {Function(!Error, ...)} callback
		* @return {Promise?}
		*/
		push(...args) {
		// Returns a promise no `callback` is given.
		if (this._workerAsync && args.length === this.worker.length \|\|
		!this._workerAsync && args.length < this.worker.length) {
		return new Promise((resolve, reject) => {
		// Add any missing arguments.
		if (!this._workerAsync) {
		while (args.length < this.worker.length - 1) {
		args.push(undefined);
		}
		}
		this.push(...args, (err, results) => {
		if (err)
		return reject(err);
		resolve(results);
		});
		});
		}
		if (this.isFull()) {
		return;
		}
		if (this.intransit < this.concurrency) {
		this.intransit++;
		this.active++;
		if (this.intransit === this.concurrency) {
		this.emit('full');
		}
		this._process(args);
		}
		else {
		this._queue.push(args);
		this.queued++;
		}
		}
		/**
		* Returns true if queue is full.
		*
		* @return {boolean}
		*/
		isFull() {
		return this.maxQueued === this.queued;
		}
		/**
		* Starts a task
		*
		* @param {Array.<Object>} args
		*/
		_process(args) {
		return __awaiter(this, void 0, void 0, function* () {
		const callback = args.pop();
		let finished = false;
		let every = ~~this.every;
		let timedOut;
		if (every) {
		timedOut = false;
		this._timers.push(setTimeout(() => {
		timedOut = true;
		this._timers.shift();
		if (finished) {
		this._next();
		}
		}, every));
		}
		else {
		timedOut = true;
		}
		// If `timeout` option is set, set a timeout to check the task doesn't lag.
		let taskTimedOut = false;
		let callbackCalled = false;
		let timeout = ~~this.timeout;
		let tid;
		const taskCallback = (err, result) => {
		// If this task has timed out, and the callback is called again
		// from the worker, ignore it.
		if (!taskTimedOut) {
		clearTimeout(tid);
		}
		else {
		return;
		}
		// Check that this callback is only called once.
		if (callbackCalled && !taskTimedOut) {
		throw Error('Callback from worker should only be called once');
		}
		callbackCalled = true;
		this.finished++;
		this.active--;
		callback(err, result);
		finished = true;
		if (timedOut) {
		this._next();
		}
		};
		if (timeout) {
		tid = setTimeout(() => {
		const err = new TimeQueue.TaskError('Task timed out');
		err.args = args;
		taskCallback(err);
		taskTimedOut = true;
		}, timeout);
		}
		// Call the worker.
		if (this._workerAsync) {
		try {
		taskCallback(null, yield this.worker(...args));
		}
		catch (err) {
		taskCallback(err);
		}
		}
		else {
		// Add custom callback to args.
		const args2 = args.slice();
		args2.push(taskCallback);
		this.worker(...args2);
		}
		});
		}
		/**
		* Called when a task finishes. Looks at the queue and processes the next
		* waiting task.
		*/
		_next() {
		if (this.intransit <= this.concurrency && this._queue.length) {
		this.queued--;
		this.active++;
		this._process(this._queue.shift());
		if (this._queue.length === 0) {
		this.emit('empty');
		}
		}
		else if (--this.intransit === 0) {
		this.emit('drain');
		}
		}
		/**
		* Empties the queue and kills the timers.
		* Active tasks will still be completed.
		*/
		die() {
		this._queue = [];
		this._timers.forEach(clearTimeout);
		this._timers = [];
		this.intransit = 0;
		this.active = 0;
		this.queued = 0;
		}
		}
		TimeQueue.TaskError = class TaskError extends Error {
		};
		module.exports = TimeQueue;
		//# sourceMappingURL=index.js.map

+1

dist/index.js.map

+14

-7

package.json

		@@ -9,3 +9,3 @@ {
		],
		"version": "2.0.0",
		"version": "2.0.1",
		"repository": {
		@@ -16,8 +16,10 @@ "type": "git",
		"author": "fent (https://github.com/fent)",
		"main": "./lib/index.js",
		"main": "./dist/index.js",
		"types": "./dist/index.d.ts",
		"files": [
		"lib"
		"dist"
		],
		"scripts": {
		"test": "nyc --reporter=lcov --reporter=text-summary mocha test/*-test.js"
		"build": "tsc -p tsconfig.build.json",
		"test": "nyc --extension .ts --reporter=lcov --reporter=text-summary mocha -- --require ts-node/register test/*-test.ts"
		},
		@@ -28,10 +30,15 @@ "directories": {
		"devDependencies": {
		"@types/mocha": "^5.2.7",
		"@types/node": "^12.12.16",
		"@types/sinon": "^7.5.1",
		"mocha": "^6.2.0",
		"nyc": "^14.1.1",
		"sinon": "^7.2.3"
		"nyc": "^15.0.0",
		"sinon": "^8.0.0",
		"ts-node": "^8.5.4",
		"typescript": "^3.7.3"
		},
		"engines": {
		"node": ">=6"
		"node": ">=10"
		},
		"license": "MIT"
		}

+1

-1

README.md

		# timequeue.js

		A queue with custom concurrency and time limits. Inspired by [caolan/async#queue](https://github.com/caolan/async#queue), but also with variable number of arguments in the worker, events, and with optional time limits.
		A queue with custom concurrency and time limits. Inspired by [async/queue](https://caolan.github.io/async/v3/docs.html#queue), but also with variable number of arguments in the worker, events, and with optional time limits.

		@@ -5,0 +5,0 @@ [![Dependency Status](https://david-dm.org/fent/timequeue.js.svg)](https://david-dm.org/fent/timequeue.js)

-207

lib/index.js

		const EventEmitter = require('events').EventEmitter;


		module.exports = class TimeQueue extends EventEmitter {
		/**
		* @constructor
		* @extends {EventEmitter}
		* @param {Function(..., Function(!Error, ...)} worker
		* @param {Object} options
		* @param {number} options.concurrency
		* @param {number} options.time
		*/
		constructor(worker, options) {
		super();

		this.worker = worker;
		this._workerAsync = worker.constructor.name == 'AsyncFunction';
		options = options \|\| {};
		this.concurrency = options.concurrency \|\| 1;
		this.every = options.every \|\| 0;
		this.maxQueued = options.maxQueued \|\| Infinity;
		this.timeout = options.timeout \|\| 0;
		this._queue = [];
		this._timers = [];

		// How many tasks are currently active.
		TimeQueue.prototype.active = 0;

		// How many tasks are still being waited on,
		// in case the `every` option was used.
		TimeQueue.prototype.intransit = 0;

		// How many tasks are in the queue.
		TimeQueue.prototype.queued = 0;

		// How many tasks have finished.
		TimeQueue.prototype.finished = 0;
		}


		/**
		* Pushes a task onto the queue.
		*
		* @param {Object} ...args
		* @param {Function(!Error, ...)} callback
		* @return {Promise?}
		*/
		push(...args) {
		// Returns a promise no `callback` is given.
		if (this._workerAsync && args.length === this.worker.length \|\|
		!this._workerAsync && args.length < this.worker.length) {
		return new Promise((resolve, reject) => {
		// Add any missing arguments.
		if (!this._workerAsync) {
		while (args.length < this.worker.length - 1) {
		args.push(undefined);
		}
		}
		this.push(...args, (err, results) => {
		if (err) return reject(err);
		resolve(results);
		});
		});
		}

		if (this.isFull()) {
		return;
		}

		if (this.intransit < this.concurrency) {
		this.intransit++;
		this.active++;
		if (this.intransit === this.concurrency) {
		this.emit('full');
		}
		this._process(args);
		} else {
		this._queue.push(args);
		this.queued++;
		}
		}


		/**
		* Returns true if queue is full.
		*
		* @return {boolean}
		*/
		isFull() {
		return this.maxQueued === this.queued;
		}


		/**
		* Starts a task
		*
		* @param {Array.<Object>} args
		*/
		async _process(args) {
		const callback = args.pop();
		let finished = false;
		let every = ~~this.every;
		let timedOut;

		if (every) {
		timedOut = false;

		this._timers.push(setTimeout(() => {
		timedOut = true;
		this._timers.shift();
		if (finished) {
		this._next();
		}
		}, every));

		} else {
		timedOut = true;
		}

		// If `timeout` option is set, set a timeout to check the task doesn't lag.
		let taskTimedOut = false;
		let callbackCalled = false;
		let timeout = ~~this.timeout;
		let tid;

		const taskCallback = (err, result) => {
		// If this task has timed out, and the callback is called again
		// from the worker, ignore it.
		if (!taskTimedOut) {
		clearTimeout(tid);
		} else {
		return;
		}

		// Check that this callback is only called once.
		if (callbackCalled && !taskTimedOut) {
		throw Error('Callback from worker should only be called once');
		}
		callbackCalled = true;
		this.finished++;
		this.active--;
		callback(err, result);

		finished = true;
		if (timedOut) {
		this._next();
		}
		};

		if (timeout) {
		tid = setTimeout(() => {
		const err = Error('Task timed out');
		err.args = args;
		taskCallback(err);
		taskTimedOut = true;
		}, timeout);
		}

		// Call the worker.
		if (this._workerAsync) {
		try {
		taskCallback(null, await this.worker(...args));
		} catch (err) {
		taskCallback(err);
		}
		} else {
		// Add custom callback to args.
		const args2 = args.slice();
		args2.push(taskCallback);
		this.worker(...args2);
		}
		}


		/**
		* Called when a task finishes. Looks at the queue and processes the next
		* waiting task.
		*/
		_next() {
		if (this.intransit <= this.concurrency && this._queue.length) {
		this.queued--;
		this.active++;
		this._process(this._queue.shift());

		if (this._queue.length === 0) {
		this.emit('empty');
		}

		} else if (--this.intransit === 0) {
		this.emit('drain');
		}
		}


		/**
		* Empties the queue and kills the timers.
		* Active tasks will still be completed.
		*/
		die() {
		this._queue = [];
		this._timers.forEach(clearTimeout);
		this._timers = [];
		this.intransit = 0;
		this.active = 0;
		this.queued = 0;
		}
		};

timequeue - npm Package Compare versions

New alerts

Fixed alerts

Improved metrics

Worsened metrics