queue

A job is a function that follows the signature (next) => { ... }.

Individual Job Timeouts: While the Queue has a default timeout, you can specify a unique timeout for an individual job by attaching a timeout property to the function object itself. The queue will prioritize the job's own timeout over the global queue setting.

const fastJob = (next) => next(null, 'fast');
fastJob.timeout = 100; // This job times out in 100ms

const slowJob = (next) => setTimeout(() => next(null, 'slow'), 5000);
// slowJob will use the Queue's default timeout

q.push(fastJob, slowJob);

When instantiating a Queue, you can provide an Options object to control its behavior:

• concurrency (number): The maximum number of jobs the queue should process concurrently. Defaults to Infinity.
• timeout (number): Milliseconds to wait for a job to execute its callback. Defaults to 0.
• autostart (boolean): If true, ensures the queue is always running if jobs are available. Useful for concurrency control. Defaults to false.
• results (any[]): An array to set job callback arguments on. Defaults to null.

const q = new Queue({
  concurrency: 5,
  timeout: 1000,
  autostart: true,
  results: []
});

Use these methods to control the lifecycle of the queue:

• stop(): Sets running to false. This prevents new jobs from being pulled from the queue, but does not cancel currently executing jobs.
• end(error): Immediately stops the queue, clears all pending timers, empties the jobs array, and triggers the end event with the provided error.

Jobs are functions that accept a next callback as their first argument. You can add multiple jobs at once using the following methods:

• push(...workers): Adds jobs to the end of the queue.
• unshift(...workers): Adds jobs to the beginning of the queue.
• splice(start, deleteCount, ...workers): Replaces or inserts jobs at a specific index.

If autostart was set to true during initialization, these methods will automatically trigger the queue processing.

// A job is a function: (next) => { ... }
const job = (next) => {
  setTimeout(() => next(null, 'done'), 100);
};

q.push(job, job, job);

The start([callback]) method begins processing the jobs in the queue.

• If a callback is provided, it will be invoked when the queue finishes (the end event), receiving (error, results).
• If no callback is provided, start() returns a Promise that resolves with the results or rejects with an error when the queue ends.

Note: Calling start() while the queue is already running will throw an Error('already started').

// Using a callback
q.start((err, results) => {
  if (err) console.error('Queue failed:', err);
  else console.log('Results:', results);
});

// Or using the returned Promise
const results = await q.start();

The Queue class provides methods to control its execution state:

• start([callback]): Starts the queue. You can provide a callback that is called when the queue empties or an error occurs. It can also be called without arguments to return a Promise that resolves with { error?: Error, results?: any[] | null }.
• stop(): Stops the queue.
• end([error]): Stops and empties the queue immediately. If an error is provided, it is passed to the start callback if one was supplied.

// Using a callback
queue.start((err, results) => {
  if (err) console.error(err);
  console.log(results);
});

// Using a Promise
const { error, results } = await queue.start();

// Stopping and emptying
queue.end();

Create a new Queue instance by passing an options object. The Queue manages asynchronous function execution with controlled concurrency.

Options:

• concurrency (Number): The maximum number of jobs to run simultaneously. Defaults to Infinity.
• timeout (Number): The default timeout in milliseconds for each job. Defaults to 0 (no timeout).
• autostart (Boolean): If true, the queue will automatically call _start() whenever jobs are added via push, unshift, or splice. Defaults to false.
• results (Array|null): An array to collect results from completed jobs. If provided, results are stored at the index corresponding to the job's order of completion.

import Queue from 'jessetane/queue';

const q = new Queue({
  concurrency: 2,
  timeout: 5000,
  autostart: true,
  results: []
});

Jobs are represented by QueueWorker objects. You can add them to the queue using several methods:

• push(...workers: QueueWorker[]): Adds one or more workers to the end of the queue. Returns the new length.
• unshift(...workers: QueueWorker[]): Adds one or more workers to the front of the queue. Returns the new length.
• splice(start, deleteCount, ...workers): Adds and/or removes elements from the queue at a specific index.
• pop(): Removes and returns the last element.
• shift(): Removes and returns the first element.

// A QueueWorker is a function that accepts an optional callback or returns a Promise
const worker: QueueWorker = (cb) => {
  // do work
  if (cb) cb(null, { data: 'success' });
};

queue.push(worker);
queue.unshift(worker);

The Queue instance provides several methods to interact with the internal jobs array. Note that some methods like slice return the Queue instance itself to allow chaining, while others return standard array results.

// Returns the job at the end of the queue
q.pop();

// Returns the job at the beginning of the queue
q.shift();

// Returns the index of a job
q.indexOf(searchElement, fromIndex);

// Returns the last index of a job
q.lastIndexOf(searchElement, fromIndex);

// Removes/replaces jobs and returns the Queue instance (chainable)
q.slice(start, end);

// Reverses the jobs array and returns the Queue instance (chainable)
q.reverse();

// Returns the total number of jobs (pending + queued)
q.length;

A QueueWorker is the unit of work processed by the queue. It must be a function that either:

1. Accepts an optional QueueWorkerCallback (which receives (error?: Error, data?: Object) => void).
2. Returns a Promise<any>.

Workers can optionally include a timeout property to override the global queue timeout and a promise property to track the worker's execution.

interface QueueWorker {
  (callback?: QueueWorkerCallback): void | Promise<any>;
  timeout?: number;
  promise?: Promise<any>;
}

// Example: Callback style
const worker1: QueueWorker = (cb) => {
  setTimeout(() => cb(null, { id: 1 }), 100);
};

// Example: Promise style
const worker2: QueueWorker = async () => {
  return { id: 2 };
};