henrygd/queue

To use the queue, create a new instance using newQueue(concurrency) and add async functions or promise-returning functions using the add method. The add method returns a promise that resolves with the same value as the supplied function. You can use queue.done() to wait for all tasks in the queue to complete.

import { newQueue } from '@henrygd/queue'

// create a new queue with a concurrency of 2
const queue = newQueue(2)

const pokemon = ['ditto', 'hitmonlee', 'pidgeot', 'poliwhirl', 'golem', 'charizard']

for (const name of pokemon) {
    queue.add(async () => {
        const res = await fetch(`https://pokeapi.co/api/v2/pokemon/${name}`)
        const json = await res.json()
        console.log(`${json.name}: ${json.height * 10}cm | ${json.weight / 10}kg`)
    })
}

console.log('running')
await queue.done()
console.log('done')

If your application requires support for Node.js AsyncLocalStorage, import the queue from the specific async-storage entry point instead of the default one:

import { ... } from '@henrygd/queue/async-storage'

For APIs with rate limits, use the @henrygd/queue/rl entry point. This allows you to limit both concurrency (simultaneous tasks) and the rate (tasks started within a specific time window).

Use newQueue(concurrency, rate?, interval?) where:

• concurrency: Maximum number of tasks running simultaneously.
• rate: Maximum number of tasks that can start within the interval (optional).
• interval: Time window in milliseconds for rate limiting (optional).

import { newQueue } from '@henrygd/queue/rl'

// max 10 concurrent requests, but only 3 can start per second
const queue = newQueue(10, 3, 1000)

const start = Date.now()

for (let i = 1; i <= 10; i++) {
   queue.add(async () => console.log(`Task ${i} started at ${Date.now() - start}ms`))
}

await queue.done()

The queue.all method behaves similarly to Promise.all but applies the queue's concurrency control. It accepts an array containing either wrapper functions (that return promises) or existing promises. Note that existing promises start executing immediately upon creation; only the wrapper functions are subject to the queue's concurrency limits.

import { newQueue } from "@henrygd/queue";

const queue = newQueue(2);

const tasks = ["ditto", "hitmonlee", "pidgeot", "poliwhirl"].map(
  (name) => async () => {
    const res = await fetch(`https://pokeapi.co/api/v2/pokemon/${name}`);
    return await res.json();
  }
);

// Process all tasks concurrently (limited by queue concurrency) and wait for all to complete
const results = await queue.all(tasks);
console.log(results);

The following methods are available on a queue instance:

• add<T>(promiseFunction: () => PromiseLike<T>): Promise<T>: Adds an async function or promise wrapper to the queue.
• all<T>(promiseFunctions: Array<PromiseLike<T> | (() => PromiseLike<T>)>): Promise<T[]>: Adds promises or wrappers and resolves like Promise.all.
• done(): Promise<void>: Returns a promise that resolves when the queue is empty.
• clear(): void: Empties the queue. Pending promises are rejected, but active promises continue.
• active(): number: Returns the number of promises currently running.
• size(): number: Returns the total number of promises currently in the queue.

Calling clear() will immediately reject all pending tasks in the queue with an Error('Queue cleared'). Any tasks that are already active (currently executing) will continue to run until they resolve or reject naturally. The size() of the queue will be updated to reflect only the remaining active tasks.

const queue = newQueue(1);

queue.add(() => new Promise(res => setTimeout(res, 1000))); // Active
queue.add(() => Promise.resolve('pending')); // Pending

queue.clear(); 
// The first task continues running.
// The second task rejects with Error('Queue cleared').

Use newQueue(concurrency) to initialize a new queue instance. The concurrency parameter defines the maximum number of operations allowed to run simultaneously. This is useful for controlling resource usage when executing many asynchronous tasks.

import { newQueue } from '@henrygd/queue';

const queue = newQueue(2); // Max 2 concurrent operations