better-queue

Initialize a new queue by passing a processing function to the Queue constructor. The function receives the task input and a callback cb to signal completion. Use q.push(data) to add tasks to the queue.

var Queue = require('better-queue');

var q = new Queue(function (input, cb) {
  // Some processing here ...
  cb(null, result);
})

q.push(1)
q.push({ x: 1 })

To use Better Queue in a TypeScript project, install the type definitions from Definitely Typed:

npm install --save @types/better-queue

import Queue = require('better-queue')

const q: Queue = new Queue(() => {});

To use Better Queue in a browser environment via Webpack, you must explicitly provide a store (the default in-memory store is not automatically provided). It is recommended to use better-queue-memory.

import Queue = require('better-queue')
import MemoryStore = require('better-queue-memory')

var q = new Queue(function (input, cb) {
  // ... processing ...
  cb(null, result);
}, {
    store: new MemoryStore(),
  });

To use Better Queue in your project, install it using npm:

npm install --save better-queue

Manage task failures and execution limits using these options:

• maxRetries: Number of times a failed task should be retried.
• retryDelay: Milliseconds to wait before retrying a failed task.
• maxTimeout: Maximum time (ms) allowed for a task to complete before it is aborted with an error.

// Retry 10 times with a 1s delay
var q = new Queue(fn, { maxRetries: 10, retryDelay: 1000 });

// Abort task if it takes longer than 2 seconds
var q = new Queue(fn, { maxTimeout: 2000 });

Better Queue supports persistent storage via SQL. You can configure these by passing a store object in the Queue options.

SQLite

Requires npm install sqlite3 and better-queue-sql or better-queue-sqlite.

PostgreSQL

Requires npm install pg.

Note: Ensure the dialect and connection parameters (host, port, etc.) are correctly provided.

// SQLite
var q = new Queue(fn, {
  store: {
    type: 'sql',
    dialect: 'sqlite',
    path: '/path/to/sqlite/file'
  }
});

// PostgreSQL
var q = new Queue(fn, {
  store: {
    type: 'sql',
    dialect: 'postgres',
    host: 'localhost',
    port: 5432,
    username: 'username',
    password: 'password',
    dbname: 'template1',
    tableName: 'tasks'
  }
});

You can control how many tasks run simultaneously and the order in which they are processed using the options object in the Queue constructor:

• concurrent: The number of tasks to process at the same time (default is 1).
• filo: Set to true to turn the queue into a stack (First-In-Last-Out).

// Run 3 tasks at a time
var q = new Queue(fn, { concurrent: 3 });

// Process items in FILO order
var q = new Queue(fn, { filo: true });

Fine-tune when tasks are processed using timing options:

• batchDelay: Wait X ms before processing a batch (useful for 'timed cargo').
• batchDelayTimeout: Force processing a batch if no new tasks have been added for X ms.
• afterProcessDelay: Delay between the completion of one task and the start of the next.
• precondition: A function that must return true before the queue processes the next batch. If it returns false, the queue waits and retries based on preconditionRetryTimeout.

// Batch processing with delays
var q = new Queue(fn, {
  batchSize: 50,
  batchDelay: 5000,
  batchDelayTimeout: 1000
});

// Using preconditions (e.g., checking internet connectivity)
var q = new Queue(fn, {
  precondition: function (cb) {
    isOnline(function (err, ok) {
      cb(null, ok);
    });
  },
  preconditionRetryTimeout: 10000
});

Use the filter and priority options to control how tasks enter and move through the queue:

• filter: A function that can transform input or reject it (by calling the callback with an error). Useful for validation or pre-processing.
• priority: A function that assigns a numeric priority to a task. Higher numbers are processed first.

// Filtering and transforming input
var greeter = new Queue(function (name, cb) {
  console.log("Hello, %s!", name);
  cb();
}, {
  filter: function (input, cb) {
    if (input === 'Bob') return cb('not_allowed');
    return cb(null, input.toUpperCase());
  }
});

// Prioritizing tasks
var q = new Queue(fn, {
  priority: function (name, cb) {
    if (name === "Steve") return cb(null, 10);
    cb(null, 1);
  }
});

To use a custom store, implement a set of required functions and either pass the store to the Queue constructor or use the queue.use(store) method.

Required Store Interface:

• connect(cb): Connect to your storage backend.
• getRunningTasks(cb): Returns a map of running tasks (lockId => taskIds).
• getTask(taskId, cb): Retrieves a specific task.
• putTask(taskId, task, priority, cb): Saves a task with a given priority.
• takeFirstN(n, cb): Removes the first n items (sorted by priority and age).
• takeLastN(n, cb): Removes the last n items (sorted by priority and recency).

var q = new Queue(fn, { store: myStore });
// OR
q.use(myStore);

To process multiple tasks at once, set the batchSize option. The processing function will then receive an array (the batch) instead of a single task.

var ages = new Queue(function (batch, cb) {
  // batch is an array of tasks
  // e.g., [ { id: 'steve', age: 21 }, { id: 'john', age: 34 } ]
  cb();
}, { batchSize: 3 });

ages.push({ id: 'steve', age: 21 });
ages.push({ id: 'john', age: 34 });
ages.push({ id: 'joe', age: 18 });
ages.push({ id: 'mary', age: 23 });