extra-promise

extra-promise

Utilities for JavaScript Promise and async functions.

Install

npm install --save extra-promise
# or
yarn add extra-promise

API

interface INonBlockingChannel<T> {
  send(value: T): void
  receive(): AsyncIterable<T>
  close: () => void
}

interface IBlockingChannel<T> {
  send(value: T): Promise<void>
  receive(): AsyncIterable<T>
  close: () => void
}

interface IDeferred<T> {
  resolve(value: T): void
  reject(reason: unknown): void
}

functions

isPromise

function isPromise<T>(val: unknown): val is Promise<T>
function isntPromise<T>(val: T): val is Exclude<T, Promise<unknown>>

isPromiseLike

function isPromiseLike<T>(val: unknown): val is PromiseLike<T>
function isntPromiseLike<T>(val: T): val is Exclude<T, PromiseLike<unknown>>

delay

function delay(timeout: number): Promise<void>

A simple wrapper for setTimeout.

timeout

function timeout(ms: number): Promise<never>

It throws a TimeoutError after ms milliseconds.

try {
  result = await Promise.race([
    fetchData()
  , timeout(5000)
  ])
} catch (e) {
  if (e instanceof TimeoutError) ...
}

pad

function pad<T>(ms: number, fn: () => Awaitable<T>): Promise<T>

Run a function, but wait at least ms milliseconds before returning.

parallel

function parallel(
  tasks: Iterable<() => Awaitable<unknown>>
, concurrency: number = Infinity
): Promise<void>

Perform tasks in parallel.

The value range of concurrency is [1, Infinity]. Invalid values will throw Error.

parallelAsync

function parallelAsync(
  tasks: AsyncIterable<() => Awaitable<unknown>>
, concurrency: number // concurrency must be finite number
): Promise<void>

Same as parallel, but tasks is an AsyncIterable.

series

function series(
  tasks: Iterable<() => Awaitable<unknown>>
       | AsyncIterable<() => Awaitable<unknown>>
): Promise<void>

Perform tasks in order. Equivalent to parallel(tasks, 1).

waterfall

function waterfall<T>(
  tasks: Iterable<(result: unknown) => Awatiable<unknown>>
       | AsyncIterable<(result: unknown) => Awaitable<unknown>>
): Promise<T | undefined>

Perform tasks in order, the return value of the previous task will become the parameter of the next task. If tasks is empty, return Promise<undefined>.

each

function each(
  iterable: Iterable<T>
, fn: (element: T, i: number) => Awaitable<unknown>
, concurrency: number = Infinity
): Promise<void>

The async each operator for Iterable.

The value range of concurrency is [1, Infinity]. Invalid values will throw Error.

eachAsync

function eachAsync<T>(
  iterable: AsyncIterable<T>
, fn: (element: T, i: number) => Awaitable<unknown>
, concurrency: number // concurrency must be finite number
): Promise<void>

Same as each, but iterable is an AsyncIterable.

map

function map<T, U>(
  iterable: Iterable<T>
, fn: (element: T, i: number) => Awaitable<U>
, concurrency: number = Infinity
): Promise<U[]>

The async map operator for Iterable.

The value range of concurrency is [1, Infinity]. Invalid values will throw Error.

mapAsync

export function mapAsync<T, U>(
  iterable: AsyncIterable<T>
, fn: (element: T, i: number) => Awaitable<U>
, concurrency: number // concurrency must be finite number
): Promise<U[]>

Same as map, but iterable is an AsyncIterable.

filter

function filter<T, U = T>(
  iterable: Iterable<T>
, fn: (element: T, i: number) => Awaitable<boolean>
, concurrency: number = Infinity
): Promise<U[]>

The async filter operator for Iterable.

The value range of concurrency is [1, Infinity]. Invalid values will throw Error.

filterAsync

function filterAsync<T, U = T>(
  iterable: AsyncIterable<T>
, fn: (element: T, i: number) => Awaitable<boolean>
, concurrency: number // concurrency must be finite number
): Promise<U[]>

Same as filter, but iterable is an AsyncIterable.

all

function all<T extends { [key: string]: PromiseLike<unknown> }>(
  obj: T
): Promise<{ [Key in keyof T]: UnpackedPromiseLike<T[Key]> }>

It is similar to Promise.all, but the first parameter is an object.

const { task1, task2 } = await all({
  task1: invokeTask1()
, task2: invokeTask2()
})

promisify

type Callback<T> = (err: any, result?: T) => void

function promisify<Result, Args extends any[] = unknown[]>(
  fn: (...args: [...args: Args, callback?: Callback<Result>]) => unknown
): (...args: Args) => Promise<Result>

The well-known promisify function.

callbackify

type Callback<T> = (err: any, result?: T) => void

function callbackify<Result, Args extends any[] = unknown[]>(
  fn: (...args: Args) => Awaitable<Result>
): (...args: [...args: Args, callback: Callback<Result>]) => void

The callbackify function, as opposed to promisify.

asyncify

function asyncify<Args extends any[], Result, This = unknown>(
  fn: (this: This, ...args: Args) => Awaitable<Result>
): (this: This, ...args: Promisify<Args>) => Promise<Result>

Turn sync functions into async functions.

const a = 1
const b = Promise.resolve(2)

const add = (a: number, b: number) => a + b

// BAD
add(a, await b) // 3

// GOOD
const addAsync = asyncify(add) // (a: number | PromiseLike<number>, b: number | PromiseLike<number>) => Promise<number>
await addAsync(a, b) // Promise<3>

It can also be used to eliminate the call stack:

// OLD
function count(n: number, i: number = 0): number {
  if (i < n) return count(n, i + 1)
  return i
}

count(10000) // RangeError: Maximum call stack size exceeded

// NEW
const countAsync = asyncify((n: number, i: number = 0): Awaitable<number> => {
  if (i < n) return countAsync(n, i + 1)
  return i
})

await countAsync(10000) // 10000

spawn

function spawn<T>(
  num: number
, create: (id: number) => Awaitable<T>
): Promise<T[]>

A sugar for create multiple values in parallel.

The parameter id is from 1 to num.

limitConcurrencyByQueue

function limitConcurrencyByQueue<T, Args extends any[]>(
  concurrency: number
, fn: (...args: Args) => PromiseLike<T>
): (...args: Args) => Promise<T>

Limit the number of concurrency, calls that exceed the number of concurrency will be delayed in order.

reusePendingPromises

type VerboseResult<T> = [value: T, isReuse: boolean]

interface IReusePendingPromisesOptions {
  verbose?: true
}

function reusePendingPromises<T, Args extends any[]>(
  fn: (...args: Args) => PromiseLike<T>
, options: IReusePendingPromiseOptions & { verbose: true }
): (...args: Args) => Promise<VerboseResult<T>>
function reusePendingPromises<T, Args extends any[]>(
  fn: (...args: Args) => PromiseLike<T>
, options: IReusePendingPromiseOptions & { verbose: false }
): (...args: Args) => Promise<T>
function reusePendingPromises<T, Args extends any[]>(
  fn: (...args: Args) => PromiseLike<T>
, options: Omit<IReusePendingPromiseOptions, 'verbose'>
): (...args: Args) => Promise<T>
function reusePendingPromises<T, Args extends any[]>(
  fn: (...args: Args) => PromiseLike<T>
): (...args: Args) => Promise<T>

Returns a function that will return the same Promise for calls with the same parameters if the Promise is pending.

Classes

StatefulPromise

enum StatefulPromiseState {
  Pending = 'pending'
, Fulfilled = 'fulfilled'
, Rejected = 'rejected'
}

class StatefulPromise<T> extends Promise<T> {
  static from<T>(promise: PromiseLike<T>): ExtraPromise<T> 

  get state(): ExtraPromiseState

  constructor(
    executor: (
      resolve: (value: T) => void
    , reject: (reason: any) => void
    ) => void
  )

  isPending(): boolean
  isFulfilled(): boolean
  isRejected(): boolean
}

A subclass of Promise used for testing, helps you understand the state of Promise.

Channel

class Channel<T> implements IBlockingChannel<T>

Implement MPMC(multi-producer, multi-consumer) FIFO queue communication with Promise and AsyncIterable.

• send Send value to the channel, block until data is taken out by the consumer.
• receive Receive value from the channel.
• close Close the channel.

If the channel closed, send and receive will throw ChannelClosedError. AsyncIterator that have already been created do not throw ChannelClosedError, but return { done: true }.

const chan = new Channel<string>()
queueMicrotask(() => {
  await chan.send('hello')
  await chan.send('world')
})
for await (const value of chan.receive()) {
  console.log(value)
}

BufferedChannel

class BufferedChannel<T> implements IBlockingChannel<T> {
  constructor(bufferSize: number)
}

Implement MPMC(multi-producer, multi-consumer) FIFO queue communication with Promise and AsyncIterable. When the amount of data sent exceeds bufferSize, send will block until data in buffer is taken out by the consumer.

• send Send value to the channel. If the buffer is full, block.
• receive Receive value from the channel.
• close Close the channel.

If the channel closed, send and receive will throw ChannelClosedError. AsyncIterator that have already been created do not throw ChannelClosedError, but return { done: true }.

const chan = new BufferedChannel<string>(1)

queueMicrotask(() => {
  await chan.send('hello')
  await chan.send('world')
})

for await (const value of chan.receive()) {
  console.log(value)
}

UnlimitedChannel

class UnlimitedChannel<T> implements INonBlockingChannel<T>

Implement MPMC(multi-producer, multi-consumer) FIFO queue communication with Promise and AsyncIterable.

UnlimitedChannel return a tuple includes three channel functions:

• send Send value to the channel. There is no size limit on the buffer, all sending will return immediately.
• receive Receive value from the channel.
• close Close the channel.

If the channel closed, send and receive will throw ChannelClosedError. AsyncIterator that have already been created do not throw ChannelClosedError, but return { done: true }.

const chan = new UnlimitedChannel<string>()

queueMicrotask(() => {
  chan.send('hello')
  chan.send('world')
})

for await (const value of chan.receive()) {
  console.log(value)
}

Deferred

class Deferred<T> implements PromiseLike<T>, IDeferred<T>

Deferred is a Promise that separates resolve() and reject() from the constructor.

MutableDeferred

class MutableDeferred<T> implements PromiseLike<T>, IDefrred<T>

MutableDeferred is similar to Deferred, but its resolve() and reject() can be called multiple times to change the value.

const deferred = new MutableDeferred()
deferred.resolve(1)
deferred.resolve(2)

await deferred // resolved(2)

ReusableDeferred

class ReusableDeferred<T> implements PromiseLike<T>, IDeferred<T>

ReusableDeferred is similar to MutableDeferred, but its internal Deferred will be overwritten with a new pending Deferred after each call.

const deferred = new ReusableDeferred()
deferred.resolve(1)
queueMicrotask(() => deferred.resolve(2))

await deferred // pending, resolved(2)

DeferredGroup

class DeferredGroup<T> implements IDeferred<T> {
  add(deferred: IDeferred<T>): void
  remove(deferred: IDeferred<T>): void
  clear(): void
}

LazyPromise

class LazyPromise<T> implements PromiseLike<T> {
  then: PromiseLike<T>['then']

  constructor(
    executor: (resolve: (value: T) => void
  , reject: (reason: any) => void) => void
  )
}

LazyPromise constructor is the same as Promise.

The difference with Promise is that LazyPromise only performs executor after then method is called.

Semaphore

type Release = () => void

class Semaphore {
  constructor(count: number)

  acquire(): Promise<Release>
  acquire<T>(handler: () => Awaitable<T>): Promise<T>
}

Mutex

type Release = () => void

class Mutex extends Semaphore {
  acquire(): Promise<Release>
  acquire<T>(handler: () => Awaitable<T>): Promise<T>
}

DebounceMicrotask

class DebounceMicrotask {
  queue(fn: () => void): void
  cancel(fn: () => void): boolean
}

queue can create a microtask, if the microtask is not executed, multiple calls will only queue it once.

cancel can cancel a microtask before it is executed.

DebounceMacrotask

class DebounceMacrotask {
  queue(fn: () => void): void
  cancel(fn: () => void): boolean
}

queue can create a macrotask, if the macrotask is not executed, multiple calls will only queue it once.

cancel can cancel a macrotask before it is executed.

TaskRunner

class TaskRunner {
  constructor(concurrency: number = Infinity)

  run(task: () => Awaitable<T>): Promise<T>
  destroy(): void
}

A task runner, it will execute tasks in FIFO order.