DOCSEARCHbyAlgolia
Submit Docs

@supercharge/promise-pool

repository·main·Indexed 21 days ago

https://github.com/supercharge/promise-pool

A Node.js utility for concurrent promise processing that provides a map-like interface to process arrays of items with controlled concurrency. It features a fluent builder pattern for configuring task timeouts, custom error handling via .handleError(), and lifecycle hooks like .onTaskStarted() and .onTaskFinished(). The library supports mapping results back to source items using .useCorrespondingResults() and provides a Statistics interface to monitor active tasks and processing progress.

Tokens
4.8K
Snippets
21
Records
26
Agent score
73%

What's inside @supercharge/promise-pool

  1. Align results with source items using .useCorrespondingResults()

    main

    By default, results are returned in the order they complete. Use .useCorrespondingResults() to ensure the results array matches the index of the source items.

    When using this mode, the results array will contain mixed types for tasks that didn't succeed:

    • The actual value (for successful tasks)
    • PromisePool.notRun (for tasks that were skipped)
    • PromisePool.failed (for tasks that failed)

    Note: When using corresponding results, you must handle errors yourself via .handleError() as the default error collection behavior is bypassed.

    import { setTimeout } from 'node:timers/promises'
import { PromisePool } from '@supercharge/promise-pool'

const { results } = await PromisePool
  .for([1, 2, 3])
  .withConcurrency(5)
  .useCorrespondingResults()
  .process(async (number, index) => {
    const value = number * 2
    return await setTimeout(10 - index, value)
  })

// results will be [2, 4, 6]

// To identify failed or skipped items:
const itemsNotRun = results.filter(result => result === PromisePool.notRun)
const failedItems = results.filter(result => result === PromisePool.failed)

  3. Use corresponding results to map results back to source items

    main

    By default, process() returns an array of results in the order they completed. If you call .useCorrespondingResults(), the returned array will match the length and order of the input items.

    If a task fails or an item is not run, the corresponding index in the result array will contain one of the following symbols:

    • PromisePool.failed: Indicates the task failed.
    • PromisePool.notRun: Indicates the item was not processed.
    import { PromisePool } from '@supercharge/promise-pool';

const items = [1, 2, 3];
const results = await PromisePool.for(items)
  .useCorrespondingResults()
  .process(async (item) => {
    if (item === 2) throw new Error('Fail');
    return item * 2;
  });

// results will be [2, PromisePool.failed, 6]

  4. Basic usage of PromisePool

    main

    Use the PromisePool class with a fluent interface to process an array of items concurrently. By default, the pool uses a concurrency of 10. You can specify a custom concurrency using .withConcurrency(n).

    import { PromisePool } from '@supercharge/promise-pool'

const users = [
  { name: 'Marcus' },
  { name: 'Norman' },
  { name: 'Christian' }
]

const { results, errors } = await PromisePool
  .withConcurrency(2)
  .for(users)
  .process(async (userData, index, pool) => {
    const user = await User.createIfNotExisting(userData)
    return user
  })

  5. Manually stop the PromisePool

    main

    You can stop the processing of a pool by calling pool.stop() from within the .process() or .handleError() methods. The pool instance is provided as an argument to these callbacks.

    // Stopping from .process()
await PromisePool
  .for(users)
  .process(async (user, index, pool) => {
    if (condition) {
      return pool.stop()
    }
  })

// Stopping from .handleError()
await PromisePool
  .for(users)
  .handleError(async (error, user, pool) => {
    if (error instanceof SomethingBadHappenedError) {
      return pool.stop()
    }
  })
  .process(async (user, index, pool) => {
    // ...
  })

  6. Configure task timeouts with .withTaskTimeout()

    main

    Use .withTaskTimeout(milliseconds) to set a timeout for each individual task. If a task exceeds this duration, it is marked as failed. This does not set a timeout for the entire pool.

    import { PromisePool } from '@supercharge/promise-pool'

await PromisePool
  .for(users)
  .withTaskTimeout(2000) // 2-second timeout per task
  .process(async (user, index, pool) => {
    // processing logic
  })

  7. Custom error handling with .handleError()

    main

    Use .handleError(handler) to implement custom error logic.

    Important: If you provide a custom error handler, the promise pool does not collect errors automatically. You must collect them yourself (e.g., by pushing them to an external array).

    To stop the entire pool immediately due to an unrecoverable error, throw an error inside the handler.

    import { PromisePool } from '@supercharge/promise-pool'

try {
  const errors = []

  const { results } = await PromisePool
    .for(users)
    .withConcurrency(4)
    .handleError(async (error, user) => {
      if (error instanceof ValidationError) {
        errors.push(error) // Manual collection required
        return
      }

      if (error instanceof ThrottleError) {
        await retryUser(user)
        return
      }

      throw error // Uncaught errors stop the pool
    })
    .process(async data => {
      // processing logic
    })

  await handleCollected(errors)
  return { results }
} catch (error) {
  await handleThrown(error)
}

  8. Hook into task lifecycle with onTaskStarted and onTaskFinished

    main

    Use .onTaskStarted() and .onTaskFinished() to execute callbacks when tasks begin or complete. You can chain multiple handlers for each event. The pool instance is available in the callbacks to inspect progress.

    import { PromisePool } from '@supercharge/promise-pool'

await PromisePool
  .for(users)
  .onTaskStarted((item, pool) => {
    console.log(`Progress: ${pool.processedPercentage()}%`)
    console.log(`Active tasks: ${pool.activeTasksCount()}`)
    console.log(`Finished tasks: ${pool.processedCount()}`)
  })
  .onTaskFinished((item, pool) => {
    // You can flush processed items to free up memory
    pool.flushProcessedItems()
  })
  .process(async (user, index, pool) => {
    // processing logic
  })

  10. Handle ValidationError when input validation fails

    main
    When using @supercharge/promise-pool, if you provide invalid arguments to the pool's methods, a ValidationError will be thrown. You can catch this error specifically to distinguish between input validation failures and errors occurring within your tasks.

  11. Monitor task progress with lifecycle callbacks

    main

    Use .onTaskStarted(handler) and .onTaskFinished(handler) to execute logic whenever a task begins or completes. These are useful for progress bars or logging.

    await PromisePool.for(items)
  .onTaskStarted((item) => console.log(`Started: ${item}`))
  .onTaskFinished((item) => console.log(`Finished: ${item}`))
  .process(async (item) => {
    // ...
  });

  12. Understand the ReturnValue interface structure

    main

    When using the promise pool, the final result is returned as a ReturnValue object. This object contains two primary arrays that allow you to distinguish between successful operations and failed ones:

    • results: An array of type R[] containing the values returned by your processing function for every successful task.
    • errors: An array of PromisePoolError<T, E> objects. Each error object includes a reference to the specific item that caused the failure via the item property, which is useful for re-processing or logging.

    Note that the length of results and errors combined will correspond to the number of items processed, but they are separated by outcome.

    interface ReturnValue<T, R, E = any> {
  results: R[];
  errors: Array<PromisePoolError<T, E>>;
}