tessl/npm-supercharge--promise-pool

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

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Pending

The risk profile of this skill

Overview

Eval results

Files

Promise Pool

Name: tessl/npm-supercharge--promise-pool
Author: tessl

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;