@portable/queue

Functions

`Queue.create`

Creates a queue that runs at most concurrency tasks at once, in priority order, optionally starting no more than intervalCap tasks in any interval milliseconds.

function create(options: create.Options = {}): Queue;

Option	Type	Default	Description
`concurrency`	`number`	`Infinity`	The most tasks that run at once: a positive integer or `Infinity`.
`intervalCap`	`number`	`Infinity`	The most tasks that start in any `interval` milliseconds. Needs `interval`.
`interval`	`number`	`0`	The length of the sliding window for `intervalCap`, in milliseconds. `0` turns rate limiting off.
`signal`	`AbortSignal`		Rejects every waiting and running task with `signal.reason`, and every task added later.

import * as Queue from '@portable/queue';
 
const queue = Queue.create({ concurrency: 4 });
const responses = await Promise.all(
  urls.map((url) => queue.add(({ signal }) => fetch(url, { signal }))),
);

`Queue.limit`

Returns a function that runs the tasks passed to it, at most concurrency at once. It is the add method of a new queue. Use create to also read its size or clear it.

function limit(concurrency: number): Queue[];

import * as Queue from '@portable/queue';
 
const limit = Queue.limit(2);
const pages = await Promise.all(
  urls.map((url) => limit(() => fetch(url).then((r) => r.text()))),
);

`Queue.map`

Calls fn on every item of input, at most concurrency at once, and resolves with the results in input order. input can be an async iterable, which is read one item at a time as slots free up. Items are passed to fn as they are, without awaiting them.

function map<chunk, result>(
  input: Iterable<chunk> | AsyncIterable<chunk>,
  fn: (
    item: chunk,
    context: map.Context,
  ) => result | PromiseLike<result>,
  options: map.Options = {},
): Promise<result[]>;

Option	Type	Default	Description
`concurrency`	`number`	`Infinity`	The most calls of `fn` that run at once: a positive integer or `Infinity`.
`signal`	`AbortSignal`		Rejects with `signal.reason`, aborts the `signal` passed to `fn` and stops reading `input`.
`stopOnError`	`boolean`	`true`	With `true`, the first error rejects at once, aborts the `signal` passed to `fn` and stops reading `input`. With `false`, `map` calls `fn` on every item and then rejects with the errors as one `AggregateError`, in input order.

import * as Queue from '@portable/queue';
 
const urls = ['https://example.com/a', 'https://example.com/b'];
const pages = await Queue.map(
  urls,
  async (url, { signal }) => (await fetch(url, { signal })).text(),
  { concurrency: 4 },
);

`Queue.once`

Wraps fn so that callers share one call: while it runs, every caller gets the same promise. After it fulfills, the result is kept (unless cache is false). After it rejects, the next call calls fn again.

function once<value>(
  fn: () => value | PromiseLike<value>,
  options: once.Options = {},
): () => Promise<value>;

Option	Type	Default	Description
`cache`	`boolean`	`true`	Whether to keep the result after `fn` fulfills, so later calls return it without calling `fn`. With `false`, callers share only a call that is still running, and the next call after it settles calls `fn` again. A rejection is never kept.

import * as Queue from '@portable/queue';
 
const connect = Queue.once(() => openDatabase(url));
const db = await connect(); // concurrent and later callers get the same connection
 
const refresh = Queue.once(() => fetchToken(), { cache: false });

Types

`Queue.Context`

What a task receives when it starts.

type Context = {
  signal: AbortSignal;
};

`Queue.Queue`

A queue made by create. Where the runtime has Symbol.asyncDispose, [Symbol.asyncDispose]() does what aborting the queue's signal does, with an AbortError DOMException, and resolves once running tasks have settled, so await using queue = Queue.create() leaves no task running after the scope ends.

type Queue = AsyncDisposeMethod & {
  add: <value>(
    task: Task<value>,
    options?: Queue.AddOptions,
  ) => Promise<value>;
  idle: () => Promise<void>;
  pause: () => void;
  resume: () => void;
  clear: () => void;
  readonly size: number;
  readonly pending: number;
  readonly paused: boolean;
  concurrency: number;
};

`Queue.Task`

A function the queue calls when a slot is free. It can return a value or a promise.

type Task<value> = (context: Context) => value | PromiseLike<value>;