fastq

Fast, in memory work queue. fastq is API compatible with async.queue

Benchmarks (1 million tasks):

• setImmediate: 812ms
• fastq: 854ms
• async.queue: 1298ms
• neoAsync.queue: 1249ms

Obtained on node 12.16.1, on a dedicated server.

If you need zero-overhead series function call, check out fastseries. For zero-overhead parallel function call, check out fastparallel.

• Installation
• Usage
• API
• Licence & copyright

Install

npm i fastq --save

Usage

'use strict'

var queue = require('fastq')(worker, 1)

queue.push(42, function (err, result) {
  if (err) { throw err }
  console.log('the result is', result)
})

function worker (arg, cb) {
  cb(null, 42 * 2)
}

Setting this

'use strict'

var that = { hello: 'world' }
var queue = require('fastq')(that, worker, 1)

queue.push(42, function (err, result) {
  if (err) { throw err }
  console.log(this)
  console.log('the result is', result)
})

function worker (arg, cb) {
  console.log(this)
  cb(null, 42 * 2)
}

API

• fastqueue()
• queue#push()
• queue#unshift()
• queue#pause()
• queue#resume()
• queue#idle()
• queue#length()
• queue#getQueue()
• queue#kill()
• queue#killAndDrain()
• queue#error()
• queue#concurrency
• queue#drain
• queue#empty
• queue#saturated

---

fastqueue([that], worker, concurrency)

Creates a new queue.

Arguments:

• that, optional context of the worker function.
• worker, worker function, it would be called with that as this, if that is specified.
• concurrency, number of concurrent tasks that could be executed in parallel.

---

queue.push(task, done)

Add a task at the end of the queue. done(err, result) will be called when the task was processed.

---

queue.unshift(task, done)

Add a task at the beginning of the queue. done(err, result) will be called when the task was processed.

---

queue.pause()

Pause the processing of tasks. Currently worked tasks are not stopped.

---

queue.resume()

Resume the processing of tasks.

---

queue.idle()

Returns false if there are tasks being processed or waiting to be processed. true otherwise.

---

queue.length()

Returns the number of tasks waiting to be processed (in the queue).

---

queue.getQueue()

Returns all the tasks be processed (in the queue). Returns empty array when there are no tasks

---

queue.kill()

Removes all tasks waiting to be processed, and reset drain to an empty function.

---

queue.killAndDrain()

Same than kill but the drain function will be called before reset to empty.

---

queue.error(handler)

Set a global error handler. handler(err, task) will be called when any of the tasks return an error.

---

queue.concurrency

Property that returns the number of concurrent tasks that could be executed in parallel. It can be altered at runtime.

---

queue.drain

Function that will be called when the last item from the queue has been processed by a worker. It can be altered at runtime.

---

queue.empty

Function that will be called when the last item from the queue has been assigned to a worker. It can be altered at runtime.

---

queue.saturated

Function that will be called when the queue hits the concurrency limit. It can be altered at runtime.

License

ISC