This commit is contained in:
2022-07-18 02:50:52 +00:00
parent befd344ab0
commit 06181b34d6
8569 changed files with 818704 additions and 352705 deletions
+11
View File
@@ -0,0 +1,11 @@
version: 2
updates:
- package-ecosystem: npm
directory: "/"
schedule:
interval: daily
open-pull-requests-limit: 10
ignore:
- dependency-name: standard
versions:
- 16.0.3
+1 -1
View File
@@ -31,7 +31,7 @@ jobs:
strategy:
matrix:
node-version: [10.x, 12.x, 13.x]
node-version: [10.x, 12.x, 13.x, 14.x, 15.x, 16.x]
steps:
- uses: actions/checkout@v2
+118 -9
View File
@@ -4,8 +4,7 @@
[![npm version][npm-badge]][npm-url]
[![Dependency Status][david-badge]][david-url]
Fast, in memory work queue. `fastq` is API compatible with
[`async.queue`](https://github.com/caolan/async#queueworker-concurrency)
Fast, in memory work queue.
Benchmarks (1 million tasks):
@@ -31,12 +30,12 @@ function call, check out [fastparallel](http://npm.im/fastparallel).
`npm i fastq --save`
## Usage
## Usage (callback API)
```js
'use strict'
var queue = require('fastq')(worker, 1)
const queue = require('fastq')(worker, 1)
queue.push(42, function (err, result) {
if (err) { throw err }
@@ -44,17 +43,34 @@ queue.push(42, function (err, result) {
})
function worker (arg, cb) {
cb(null, 42 * 2)
cb(null, arg * 2)
}
```
### Setting this
## Usage (promise API)
```js
const queue = require('fastq').promise(worker, 1)
async function worker (arg) {
return arg * 2
}
async function run () {
const result = await queue.push(42)
console.log('the result is', result)
}
run()
```
### Setting "this"
```js
'use strict'
var that = { hello: 'world' }
var queue = require('fastq')(that, worker, 1)
const that = { hello: 'world' }
const queue = require('fastq')(that, worker, 1)
queue.push(42, function (err, result) {
if (err) { throw err }
@@ -64,7 +80,51 @@ queue.push(42, function (err, result) {
function worker (arg, cb) {
console.log(this)
cb(null, 42 * 2)
cb(null, arg * 2)
}
```
### Using with TypeScript (callback API)
```ts
'use strict'
import * as fastq from "fastq";
import type { queue, done } from "fastq";
type Task = {
id: number
}
const q: queue<Task> = fastq(worker, 1)
q.push({ id: 42})
function worker (arg: Task, cb: done) {
console.log(arg.id)
cb(null)
}
```
### Using with TypeScript (promise API)
```ts
'use strict'
import * as fastq from "fastq";
import type { queueAsPromised } from "fastq";
type Task = {
id: number
}
const q: queueAsPromised<Task> = fastq.promise(asyncWorker, 1)
q.push({ id: 42}).catch((err) => console.error(err))
async function asyncWorker (arg: Task): Promise<void> {
// No need for a try-catch block, fastq handles errors automatically
console.log(arg.id)
}
```
@@ -80,10 +140,12 @@ function worker (arg, cb) {
* <a href="#getQueue"><code>queue#<b>getQueue()</b></code></a>
* <a href="#kill"><code>queue#<b>kill()</b></code></a>
* <a href="#killAndDrain"><code>queue#<b>killAndDrain()</b></code></a>
* <a href="#error"><code>queue#<b>error()</b></code></a>
* <a href="#concurrency"><code>queue#<b>concurrency</b></code></a>
* <a href="#drain"><code>queue#<b>drain</b></code></a>
* <a href="#empty"><code>queue#<b>empty</b></code></a>
* <a href="#saturated"><code>queue#<b>saturated</b></code></a>
* <a href="#promise"><code>fastqueue.promise()</code></a>
-------------------------------------------------------
<a name="fastqueue"></a>
@@ -158,6 +220,13 @@ function.
Same than `kill` but the `drain` function will be called before reset to empty.
-------------------------------------------------------
<a name="error"></a>
### queue.error(handler)
Set a global error handler. `handler(err, task)` will be called
when any of the tasks return an error.
-------------------------------------------------------
<a name="concurrency"></a>
### queue.concurrency
@@ -189,6 +258,46 @@ Function that will be called when the queue hits the concurrency
limit.
It can be altered at runtime.
-------------------------------------------------------
<a name="promise"></a>
### fastqueue.promise([that], worker(arg), concurrency)
Creates a new queue with `Promise` apis. It also offers all the methods
and properties of the object returned by [`fastqueue`](#fastqueue) with the modified
[`push`](#pushPromise) and [`unshift`](#unshiftPromise) methods.
Node v10+ is required to use the promisified version.
Arguments:
* `that`, optional context of the `worker` function.
* `worker`, worker function, it would be called with `that` as `this`,
if that is specified. It MUST return a `Promise`.
* `concurrency`, number of concurrent tasks that could be executed in
parallel.
<a name="pushPromise"></a>
#### queue.push(task) => Promise
Add a task at the end of the queue. The returned `Promise` will be fulfilled (rejected)
when the task is completed successfully (unsuccessfully).
This promise could be ignored as it will not lead to a `'unhandledRejection'`.
<a name="unshiftPromise"></a>
#### queue.unshift(task) => Promise
Add a task at the beginning of the queue. The returned `Promise` will be fulfilled (rejected)
when the task is completed successfully (unsuccessfully).
This promise could be ignored as it will not lead to a `'unhandledRejection'`.
<a name="drained"></a>
#### queue.drained() => Promise
Wait for the queue to be drained. The returned `Promise` will be resolved when all tasks in the queue have been processed by a worker.
This promise could be ignored as it will not lead to a `'unhandledRejection'`.
## License
ISC
+17 -9
View File
@@ -1,15 +1,18 @@
'use strict'
var max = 1000000
var fastqueue = require('./')(worker, 1)
var async = require('async')
var neo = require('neo-async')
var asyncqueue = async.queue(worker, 1)
var neoqueue = neo.queue(worker, 1)
const max = 1000000
const fastqueue = require('./')(worker, 1)
const { promisify } = require('util')
const immediate = promisify(setImmediate)
const qPromise = require('./').promise(immediate, 1)
const async = require('async')
const neo = require('neo-async')
const asyncqueue = async.queue(worker, 1)
const neoqueue = neo.queue(worker, 1)
function bench (func, done) {
var key = max + '*' + func.name
var count = -1
const key = max + '*' + func.name
let count = -1
console.time(key)
end()
@@ -46,12 +49,17 @@ function benchSetImmediate (cb) {
worker(42, cb)
}
function benchFastQPromise (done) {
qPromise.push(42).then(function () { done() }, done)
}
function runBench (done) {
async.eachSeries([
benchSetImmediate,
benchFastQ,
benchNeoQueue,
benchAsyncQueue
benchAsyncQueue,
benchFastQPromise
], bench, done)
}
+2
View File
@@ -1,5 +1,7 @@
'use strict'
/* eslint-disable no-var */
var queue = require('./')(worker, 1)
queue.push(42, function (err, result) {
+11
View File
@@ -0,0 +1,11 @@
import { promise as queueAsPromised } from './queue.js'
/* eslint-disable */
const queue = queueAsPromised(worker, 1)
console.log('the result is', await queue.push(42))
async function worker (arg) {
return 42 * 2
}
+14 -2
View File
@@ -3,11 +3,13 @@ declare function fastq<C, T = any, R = any>(worker: fastq.worker<C, T, R>, concu
declare namespace fastq {
type worker<C, T = any, R = any> = (this: C, task: T, cb: fastq.done<R>) => void
type asyncWorker<C, T = any, R = any> = (this: C, task: T) => Promise<R>
type done<R = any> = (err: Error | null, result?: R) => void
type errorHandler<T = any> = (err: Error, task: T) => void
interface queue<T = any, R = any> {
push(task: T, done: done<R>): void
unshift(task: T, done: done<R>): void
push(task: T, done?: done<R>): void
unshift(task: T, done?: done<R>): void
pause(): any
resume(): any
idle(): boolean
@@ -15,11 +17,21 @@ declare namespace fastq {
getQueue(): T[]
kill(): any
killAndDrain(): any
error(handler: errorHandler): void
concurrency: number
drain(): any
empty: () => void
saturated: () => void
}
interface queueAsPromised<T = any, R = any> extends queue<T, R> {
push(task: T): Promise<R>
unshift(task: T): Promise<R>
drained(): Promise<void>
}
function promise<C, T = any, R = any>(context: C, worker: fastq.asyncWorker<C, T, R>, concurrency: number): fastq.queueAsPromised<T, R>
function promise<C, T = any, R = any>(worker: fastq.asyncWorker<C, T, R>, concurrency: number): fastq.queueAsPromised<T, R>
}
export = fastq
+23 -18
View File
@@ -1,32 +1,32 @@
{
"_args": [
[
"fastq@1.8.0",
"/mnt/Foxconn/Digitalent/Deverloper/liff-push_2series"
"fastq@1.13.0",
"/home/node/nuxt"
]
],
"_from": "fastq@1.8.0",
"_id": "fastq@1.8.0",
"_from": "fastq@1.13.0",
"_id": "fastq@1.13.0",
"_inBundle": false,
"_integrity": "sha512-SMIZoZdLh/fgofivvIkmknUXyPnvxRE3DhtZ5Me3Mrsk5gyPL42F0xr51TdRXskBxHfMp+07bcYzfsYEsSQA9Q==",
"_integrity": "sha512-YpkpUnK8od0o1hmeSc7UUs/eB/vIPWJYjKck2QKIzAf71Vm1AAQ3EbuZB3g2JIy+pg+ERD0vqI79KyZiB2e2Nw==",
"_location": "/fastq",
"_phantomChildren": {},
"_requested": {
"type": "version",
"registry": true,
"raw": "fastq@1.8.0",
"raw": "fastq@1.13.0",
"name": "fastq",
"escapedName": "fastq",
"rawSpec": "1.8.0",
"rawSpec": "1.13.0",
"saveSpec": null,
"fetchSpec": "1.8.0"
"fetchSpec": "1.13.0"
},
"_requiredBy": [
"/@nodelib/fs.walk"
],
"_resolved": "https://registry.npmjs.org/fastq/-/fastq-1.8.0.tgz",
"_spec": "1.8.0",
"_where": "/mnt/Foxconn/Digitalent/Deverloper/liff-push_2series",
"_resolved": "https://registry.npmjs.org/fastq/-/fastq-1.13.0.tgz",
"_spec": "1.13.0",
"_where": "/home/node/nuxt",
"author": {
"name": "Matteo Collina",
"email": "hello@matteocollina.com"
@@ -43,10 +43,10 @@
"neo-async": "^2.6.1",
"nyc": "^15.0.0",
"pre-commit": "^1.2.2",
"snazzy": "^8.0.0",
"standard": "^14.0.0",
"tape": "^4.13.2",
"typescript": "^3.8.3"
"snazzy": "^9.0.0",
"standard": "^16.0.0",
"tape": "^5.0.0",
"typescript": "^4.0.2"
},
"homepage": "https://github.com/mcollina/fastq#readme",
"keywords": [
@@ -66,13 +66,18 @@
"url": "git+https://github.com/mcollina/fastq.git"
},
"scripts": {
"coverage": "nyc --reporter=html --reporter=cobertura --reporter=text tape test/test.js",
"coverage": "nyc --reporter=html --reporter=cobertura --reporter=text tape test/test.js test/promise.js",
"legacy": "tape test/test.js",
"lint": "standard --verbose | snazzy",
"test": "npm run lint && npm run unit && npm run typescript",
"test:report": "npm run lint && npm run unit:report",
"typescript": "tsc --project ./test/tsconfig.json",
"unit": "nyc --lines 100 --branches 100 --functions 100 --check-coverage --reporter=text tape test/test.js"
"unit": "nyc --lines 100 --branches 100 --functions 100 --check-coverage --reporter=text tape test/test.js test/promise.js"
},
"version": "1.8.0"
"standard": {
"ignore": [
"example.mjs"
]
},
"version": "1.13.0"
}
+98 -1
View File
@@ -1,5 +1,7 @@
'use strict'
/* eslint-disable no-var */
var reusify = require('reusify')
function fastqueue (context, worker, concurrency) {
@@ -9,10 +11,15 @@ function fastqueue (context, worker, concurrency) {
context = null
}
if (concurrency < 1) {
throw new Error('fastqueue concurrency must be greater than 1')
}
var cache = reusify(Task)
var queueHead = null
var queueTail = null
var _running = 0
var errorHandler = null
var self = {
push: push,
@@ -29,7 +36,8 @@ function fastqueue (context, worker, concurrency) {
unshift: unshift,
empty: noop,
kill: kill,
killAndDrain: killAndDrain
killAndDrain: killAndDrain,
error: error
}
return self
@@ -86,6 +94,7 @@ function fastqueue (context, worker, concurrency) {
current.release = release
current.value = value
current.callback = done || noop
current.errorHandler = errorHandler
if (_running === self.concurrency || self.paused) {
if (queueTail) {
@@ -161,6 +170,10 @@ function fastqueue (context, worker, concurrency) {
self.drain()
self.drain = noop
}
function error (handler) {
errorHandler = handler
}
}
function noop () {}
@@ -171,16 +184,100 @@ function Task () {
this.next = null
this.release = noop
this.context = null
this.errorHandler = null
var self = this
this.worked = function worked (err, result) {
var callback = self.callback
var errorHandler = self.errorHandler
var val = self.value
self.value = null
self.callback = noop
if (self.errorHandler) {
errorHandler(err, val)
}
callback.call(self.context, err, result)
self.release(self)
}
}
function queueAsPromised (context, worker, concurrency) {
if (typeof context === 'function') {
concurrency = worker
worker = context
context = null
}
function asyncWrapper (arg, cb) {
worker.call(this, arg)
.then(function (res) {
cb(null, res)
}, cb)
}
var queue = fastqueue(context, asyncWrapper, concurrency)
var pushCb = queue.push
var unshiftCb = queue.unshift
queue.push = push
queue.unshift = unshift
queue.drained = drained
return queue
function push (value) {
var p = new Promise(function (resolve, reject) {
pushCb(value, function (err, result) {
if (err) {
reject(err)
return
}
resolve(result)
})
})
// Let's fork the promise chain to
// make the error bubble up to the user but
// not lead to a unhandledRejection
p.catch(noop)
return p
}
function unshift (value) {
var p = new Promise(function (resolve, reject) {
unshiftCb(value, function (err, result) {
if (err) {
reject(err)
return
}
resolve(result)
})
})
// Let's fork the promise chain to
// make the error bubble up to the user but
// not lead to a unhandledRejection
p.catch(noop)
return p
}
function drained () {
var previousDrain = queue.drain
var p = new Promise(function (resolve) {
queue.drain = function () {
previousDrain()
resolve()
}
})
return p
}
}
module.exports = fastqueue
module.exports.promise = queueAsPromised
+18
View File
@@ -1,4 +1,5 @@
import * as fastq from '../'
import { promise as queueAsPromised } from '../'
// Basic example
@@ -9,6 +10,8 @@ queue.push('world', (err, result) => {
console.log('the result is', result)
})
queue.push('push without cb')
queue.concurrency
queue.drain()
@@ -36,6 +39,8 @@ queue.unshift('world', (err, result) => {
console.log('the result is', result)
})
queue.unshift('unshift without cb')
function worker(task: any, cb: fastq.done) {
cb(null, 'hello ' + task)
}
@@ -61,3 +66,16 @@ genericsQueue.unshift(7, (err, done) => {
function genericsWorker(this: GenericsContext, task: number, cb: fastq.done<string>) {
cb(null, 'the meaning of life is ' + (this.base * task))
}
const queue2 = queueAsPromised(asyncWorker, 1)
async function asyncWorker(task: any) {
return 'hello ' + task
}
async function run () {
await queue.push(42)
await queue.unshift(42)
}
run()
+221
View File
@@ -0,0 +1,221 @@
'use strict'
const test = require('tape')
const buildQueue = require('../').promise
const { promisify } = require('util')
const sleep = promisify(setTimeout)
const immediate = promisify(setImmediate)
test('concurrency', function (t) {
t.plan(2)
t.throws(buildQueue.bind(null, worker, 0))
t.doesNotThrow(buildQueue.bind(null, worker, 1))
async function worker (arg) {
return true
}
})
test('worker execution', async function (t) {
const queue = buildQueue(worker, 1)
const result = await queue.push(42)
t.equal(result, true, 'result matches')
async function worker (arg) {
t.equal(arg, 42)
return true
}
})
test('limit', async function (t) {
const queue = buildQueue(worker, 1)
const [res1, res2] = await Promise.all([queue.push(10), queue.push(0)])
t.equal(res1, 10, 'the result matches')
t.equal(res2, 0, 'the result matches')
async function worker (arg) {
await sleep(arg)
return arg
}
})
test('multiple executions', async function (t) {
const queue = buildQueue(worker, 1)
const toExec = [1, 2, 3, 4, 5]
const expected = ['a', 'b', 'c', 'd', 'e']
let count = 0
await Promise.all(toExec.map(async function (task, i) {
const result = await queue.push(task)
t.equal(result, expected[i], 'the result matches')
}))
async function worker (arg) {
t.equal(arg, toExec[count], 'arg matches')
return expected[count++]
}
})
test('drained', async function (t) {
const queue = buildQueue(worker, 2)
const toExec = new Array(10).fill(10)
let count = 0
async function worker (arg) {
await sleep(arg)
count++
}
toExec.forEach(function (i) {
queue.push(i)
})
await queue.drained()
t.equal(count, toExec.length)
toExec.forEach(function (i) {
queue.push(i)
})
await queue.drained()
t.equal(count, toExec.length * 2)
})
test('drained with exception should not throw', async function (t) {
const queue = buildQueue(worker, 2)
const toExec = new Array(10).fill(10)
async function worker () {
throw new Error('foo')
}
toExec.forEach(function (i) {
queue.push(i)
})
await queue.drained()
})
test('drained with drain function', async function (t) {
let drainCalled = false
const queue = buildQueue(worker, 2)
queue.drain = function () {
drainCalled = true
}
const toExec = new Array(10).fill(10)
let count = 0
async function worker (arg) {
await sleep(arg)
count++
}
toExec.forEach(function () {
queue.push()
})
await queue.drained()
t.equal(count, toExec.length)
t.equal(drainCalled, true)
})
test('set this', async function (t) {
t.plan(1)
const that = {}
const queue = buildQueue(that, worker, 1)
await queue.push(42)
async function worker (arg) {
t.equal(this, that, 'this matches')
}
})
test('unshift', async function (t) {
const queue = buildQueue(worker, 1)
const expected = [1, 2, 3, 4]
await Promise.all([
queue.push(1),
queue.push(4),
queue.unshift(3),
queue.unshift(2)
])
t.is(expected.length, 0)
async function worker (arg) {
t.equal(expected.shift(), arg, 'tasks come in order')
}
})
test('push with worker throwing error', async function (t) {
t.plan(5)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
q.error(function (err, task) {
t.ok(err instanceof Error, 'global error handler should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
t.equal(task, 42, 'The task executed should be passed')
})
try {
await q.push(42)
} catch (err) {
t.ok(err instanceof Error, 'push callback should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
}
})
test('unshift with worker throwing error', async function (t) {
t.plan(2)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
try {
await q.unshift(42)
} catch (err) {
t.ok(err instanceof Error, 'push callback should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
}
})
test('no unhandledRejection (push)', async function (t) {
function handleRejection () {
t.fail('unhandledRejection')
}
process.once('unhandledRejection', handleRejection)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
q.push(42)
await immediate()
process.removeListener('unhandledRejection', handleRejection)
})
test('no unhandledRejection (unshift)', async function (t) {
function handleRejection () {
t.fail('unhandledRejection')
}
process.once('unhandledRejection', handleRejection)
const q = buildQueue(async function (task, cb) {
throw new Error('test error')
}, 1)
q.unshift(42)
await immediate()
process.removeListener('unhandledRejection', handleRejection)
})
+28
View File
@@ -1,8 +1,20 @@
'use strict'
/* eslint-disable no-var */
var test = require('tape')
var buildQueue = require('../')
test('concurrency', function (t) {
t.plan(2)
t.throws(buildQueue.bind(null, worker, 0))
t.doesNotThrow(buildQueue.bind(null, worker, 1))
function worker (arg, cb) {
cb(null, true)
}
})
test('worker execution', function (t) {
t.plan(3)
@@ -536,3 +548,19 @@ test('unshift without cb', function (t) {
cb()
}
})
test('push with worker throwing error', function (t) {
t.plan(5)
var q = buildQueue(function (task, cb) {
cb(new Error('test error'), null)
}, 1)
q.error(function (err, task) {
t.ok(err instanceof Error, 'global error handler should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
t.equal(task, 42, 'The task executed should be passed')
})
q.push(42, function (err) {
t.ok(err instanceof Error, 'push callback should catch the error')
t.match(err.message, /test error/, 'error message should be "test error"')
})
})
+2 -2
View File
@@ -3,9 +3,9 @@
"target": "es6",
"module": "commonjs",
"noEmit": true,
"strict": true,
"strict": true
},
"files": [
"./example.ts"
]
}
}