fastq

When creating a queue, you can optionally provide a that argument as the first parameter to fastq() or fastq.promise(). This object will be used as the this context inside your worker function.

'use strict'

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

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

function worker (arg, cb) {
  console.log(this) // Accesses { hello: 'world' }
  cb(null, arg * 2)
}

Fastq provides type definitions for both callback and promise APIs.

Callback API: Use the queue and done types.

Promise API: Use the queueAsPromised type.

// Callback API
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)
}

// Promise API
import type { queueAsPromised } from "fastq";

const qP: queueAsPromised<Task> = fastq.promise(asyncWorker, 1)

qP.push({ id: 42 }).catch((err) => console.error(err))

async function asyncWorker (arg: Task): Promise<void> {
  console.log(arg.id)
}

Install the fastq package via npm to use it as a fast, in-memory work queue.

npm i fastq --save

The standard callback API allows you to create a queue by passing a worker function and a concurrency limit. Tasks are added using queue.push(task, done), where done is a callback invoked with (err, result) once the task is processed.

'use strict'

const 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, arg * 2)
}

Creates a queue optimized for Promise-based workflows. It includes all standard queue methods plus promise-specific versions of push, unshift, and drained.

Arguments:

• that (optional): The context (this) for the worker function.
• worker (function): A function that must return a Promise.
• concurrency (number): The number of concurrent tasks.

Promise Methods:

• queue.push(task): Returns a Promise that fulfills/rejects when the task completes. This promise can be ignored without causing unhandled rejections.
• queue.unshift(task): Returns a Promise that fulfills/rejects when the task completes.
• queue.drained(): Returns a Promise that resolves when all tasks in the queue have been processed.

Control the flow of task processing:

• queue.pause(): Pauses the processing of new tasks. Tasks currently being worked on are not stopped.
• queue.resume(): Resumes task processing.
• queue.paused (Read-Only): A boolean property indicating if the queue is currently paused.

Check the status of the queue:

• queue.length(): Returns the number of tasks currently waiting in the queue.
• queue.idle(): Returns true if no tasks are being processed or waiting; false otherwise.
• queue.getQueue(): Returns an array of all tasks currently waiting in the queue.

Stop the queue and clear pending tasks:

• queue.kill(): Removes all waiting tasks and resets the drain function to an empty function.
• queue.killAndDrain(): Same as kill, but the drain function is called before the reset.

Add tasks to the queue:

• queue.push(task, done): Adds a task to the end of the queue. In the promise API, push(task) returns a Promise.
• queue.unshift(task, done): Adds a task to the beginning of the queue. In the promise API, unshift(task) returns a Promise.

Creates a new queue instance.

Arguments:

• that (optional): The context (this) for the worker function.
• worker (function): The worker function that processes tasks.
• concurrency (number): The number of tasks to execute in parallel.

Customize how the queue behaves during specific lifecycle events or errors:

• queue.error(handler): Sets a global error handler. The handler is called as handler(err, task) whenever a task fails.
• queue.drain: A function called when the last item in the queue has been processed. Can be altered at runtime.
• queue.empty: A function called when the last item in the queue has been assigned to a worker. Can be altered at runtime.
• queue.saturated: A function called when the queue reaches its concurrency limit. Can be altered at runtime.
• queue.concurrency: Returns the current concurrency limit. This value can be altered at runtime.

Set a global error handler for the queue using error(handler). This handler will be called whenever a task fails (i.e., when the worker calls worked(err, result) with a truthy err).

handler(err, value): err is the error object, and value is the original task value.

queue.error((err, val) => {
  console.error(`Task ${val} failed with: ${err.message}`);
});