Attq!

An async tiny task queue (and related utilities), with zero dependencies. Attq provides a data structure that executes an asynchronous callback sequentially on a flexible list. It is generally designed to facilitate replaying ordered events from a client to a server and comes with bells and whistles including batching, throttling, and configurable retries.

Installation

npm install --save attq

Examples

Basic Task Queue

The AsyncTaskQueue constructor takes a task handler callback, and items are added to the queue object with the .push() method. Additional options may be configured with methods such as .onError(), .batchSize(), and .throttleMs().

import { AsyncTaskQueue } from "attq";

const q = new AsyncTaskQueue<number>(
  (nums: number[]) => fetch(`/refine?macrodata=${nums.join(",")}`),
)
  // Handler will receive up to 4 items per batch.
  .batchSize(4)
  // Handler will be called at most once per 100 milliseconds.
  .throttleMs(100);

// Add items to the queue.
for (let n = 0; n < 1000; n++) {
  q.push(n);
}

await new Promise((resolve) => setTimeout(resolve, 5000));

console.log(q.size()); // Should display >=800 items remaining.

// Stop processing items if any causes an exception.
q.onError(() => {
  q.drain();
});

Retries

Rather than build retry logic into the queue itself, Attq provides a withRetry() higher-order function which can be wrapped around the task handler. (If desired, withRetry() may be used independently of the task queue as well!)

import { AsyncTaskQueue, withRetry } from "attq";

let q = new AsyncTaskQueue<number>(
  // Defaults to 6 attempts with binary exponential backoff.
  withRetry((nums) => fetch(`/refine?macrodata=${nums.join(",")}`)),
);

// To specify, for example, up to 3 attempts with a linear backoff:
q = new AsyncTaskQueue<number>(
  withRetry(
    (nums) => fetch(`/refine?macrodata=${nums.join(",")}`),
    {
      attempts: 3,
      backoffMs: (attempt) => 1000 * attempt,
    },
  ),
);