or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

core-pool-management.mderror-handling.mdindex.mdpool-control.mdprogress-tracking.md
tile.json

tessl/npm-supercharge--promise-pool

Map-like, concurrent promise processing for Node.js with configurable concurrency limits, error handling, and advanced features.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
npmpkg:npm/@supercharge/promise-pool@3.2.x

To install, run

npx @tessl/cli install tessl/npm-supercharge--promise-pool@3.2.0

index.mddocs/

Promise Pool

Promise Pool provides map-like, concurrent promise processing for Node.js with configurable concurrency limits, error handling, and advanced features. It enables controlled parallel execution of asynchronous operations while maintaining fine-grained control over execution flow, error handling, and result ordering.

Package Information

  • Package Name: @supercharge/promise-pool
  • Package Type: npm
  • Language: TypeScript
  • Installation: npm install @supercharge/promise-pool

Core Imports

import { PromisePool } from "@supercharge/promise-pool";

Default import:

import PromisePool from "@supercharge/promise-pool";

CommonJS:

const { PromisePool } = require("@supercharge/promise-pool");

Basic Usage

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;
  });

Architecture

Promise Pool is built around several key components:

  • Fluent Interface: PromisePool class providing chainable method calls for configuration
  • Executor Engine: PromisePoolExecutor handles the internal processing logic
  • Error Handling: Custom error classes and configurable error handling strategies
  • Progress Tracking: Real-time statistics and callback system for monitoring task execution
  • Result Management: Options for maintaining correspondence between source items and results
  • Concurrency Control: Configurable limits with dynamic adjustment capabilities

Capabilities

Core Pool Management

Primary promise pool functionality for creating, configuring, and executing concurrent tasks with full type safety.

class PromisePool<T, ShouldUseCorrespondingResults extends boolean = false> {
  constructor(items?: SomeIterable<T>);
  static withConcurrency(concurrency: number): PromisePool<unknown>;
  static withTaskTimeout(timeout: number): PromisePool<unknown>;
  static for<T>(items: SomeIterable<T>): PromisePool<T>;
  
  withConcurrency(concurrency: number): PromisePool<T>;
  withTaskTimeout(timeout: number): PromisePool<T>;
  for<ItemType>(items: SomeIterable<ItemType>): PromisePool<ItemType>;
  useCorrespondingResults(): PromisePool<T, true>;
  
  process<ResultType, ErrorType = any>(
    callback: ProcessHandler<T, ResultType>
  ): Promise<ReturnValue<T, ShouldUseCorrespondingResults extends true ? ResultType | symbol : ResultType, ErrorType>>;
}

Core Pool Management

Error Handling

Comprehensive error handling system with custom error handlers, error wrapping, and selective error processing.

handleError(handler: ErrorHandler<T>): PromisePool<T>;

type ErrorHandler<T> = (
  error: Error, 
  item: T, 
  pool: Stoppable & UsesConcurrency
) => Promise<void> | void;

class PromisePoolError<T, E = any> extends Error {
  item: T;
  raw: E;
  constructor(error: E, item: T);
  static createFrom<T, E = any>(error: E, item: T): PromisePoolError<T>;
}

Error Handling

Progress Tracking

Real-time progress monitoring with task lifecycle callbacks and comprehensive statistics.

onTaskStarted(handler: OnProgressCallback<T>): PromisePool<T>;
onTaskFinished(handler: OnProgressCallback<T>): PromisePool<T>;

type OnProgressCallback<T> = (
  item: T, 
  pool: Stoppable & Statistics<T> & UsesConcurrency
) => void;

interface Statistics<T> {
  activeTasksCount(): number;
  activeTaskCount(): number; // deprecated - use activeTasksCount
  processedItems(): T[];
  processedCount(): number;
  processedPercentage(): number;
}

Progress Tracking

Pool Control

Manual pool control capabilities including stopping execution and checking pool state.

interface Stoppable {
  stop(): void;
  isStopped(): boolean;
}

interface UsesConcurrency {
  useConcurrency(concurrency: number): this;
  concurrency(): number;
}

Pool Control

Types

type SomeIterable<T> = T[] | Iterable<T> | AsyncIterable<T>;

type ProcessHandler<T, R> = (
  item: T, 
  index: number, 
  pool: Stoppable & UsesConcurrency
) => Promise<R> | R;

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

// Special symbols for corresponding results
static readonly notRun: symbol;
static readonly failed: symbol;