npmUtilsupdated 08 Aug 2026

parallel-transform

Item: parallel-transform
Rating: 2.5
Author: Keyoor

parallel-transform is a CommonJS Node transform stream that allows several callback-style transformations to be in flight at once. It defaults to object mode and preserves input order by holding completed results until every earlier item finishes; `{ ordered: false }` emits values when their callbacks complete. The first argument sets the concurrency limit, while ordinary stream backpressure limits how much input is accepted. This overlaps asynchronous work such as I/O, but it does not create threads or make CPU-bound JavaScript execute in parallel.

parallel-transform · v1.2.0 · 4.0M/wk · ★ 80 · pushed 14 Jun 2024 · mafintosh/parallel-transform · docs · reviewed by Keyoor

Verdict

Keep it for proven legacy callback pipelines that need ordered asynchronous concurrency. For new code, prefer promise-native iterable processing or a small current Transform implementation with explicit abort and error semantics.

API stability4/5The constructor signature, callback worker, default object mode, ordered output, `ordered: false`, and stream-options position have remained unchanged for years. Its frozen surface is easy to understand, but it inherits old readable-stream behavior and overrides `destroy()` with a minimal close emitter rather than following current destroy and error conventions.

Docs2/5The README gives a clear ordered-concurrency example, explains the concurrency argument, documents binary mode, and names the unordered option. It omits required positive limits, callback-once discipline, null filtering, promise incompatibility, error propagation, backpressure interaction, cancellation, head-of-line blocking, TypeScript, and the exact default high-water mark.

Maintenance1/5npm lists 1.2.0 from September 2019 as the latest release, while GitHub reports the last repository push in June 2024. The repository is not archived or deprecated, but no newer package modernized its readable-stream 2 dependency, custom destroy implementation, callback-only worker, module format, declarations, or cancellation support.

Ecosystem3/5The package recorded 4,039,695 downloads in the measured week, indicating a substantial transitive footprint in older build and stream tools. Its direct community surface is small at 80 GitHub stars, and it integrates only through the standard stream shape; there are no adapters for promises, AbortSignal, worker threads, observables, or Web Streams.

Use it if

You maintain a callback-based CommonJS stream pipeline that already uses readable-stream 2 semantics
Each input starts independent asynchronous I/O and a small fixed concurrency limit improves throughput
You must preserve input order even when individual operations complete out of order
You intentionally use `null` or `undefined` callback results to filter chunks from an object stream

Skip it if

Your transform is CPU-bound: callbacks still execute on the Node event loop, so this package provides concurrency but no worker-thread parallelism
You write promise or async-function code: the API recognizes only an error-first callback and does not await a returned promise
You need active maintenance and current stream internals: 1.2.0 was published in 2019 and depends on readable-stream 2 plus custom legacy `destroy()` behavior
One slow item can block later output and memory must stay tight: ordered mode has head-of-line blocking even when many later transforms have completed
You need cancellation of in-flight work: destroying the transform makes late callbacks no-ops, but the package has no AbortSignal and does not stop requests, timers, file operations, or other work started by your callback

Setup reality

`npm install parallel-transform` adds three runtime dependencies: `readable-stream` 2, `inherits`, and `cyclist`. There are no peers, native builds, credentials, declarations, ESM entry point, or config file. Construct it with a positive concurrency integer and a callback-style worker. The implementation does not validate that number, so zero, negative, fractional, or nonnumeric values can stall or corrupt scheduling; validate user configuration before creating the stream. Object mode is on unless `objectMode: false`, and its default high-water mark is the larger of the concurrency value and 16. The concurrency limit counts work whose callback has not yet drained in input order, while the high-water mark controls buffered stream writes, so setting concurrency to a large number can also raise upstream buffering. Every worker must call its callback exactly once. Never throw asynchronously, forget a callback, both throw and call back, or call it twice. A callback error emits `error`, ends readable output, and emits `close`, but the custom destroy method does not cancel other workers. Use `stream.pipeline()` or explicit error listeners so errors do not become uncaught process events. Passing `null` or `undefined` as successful data drops that input, so those values cannot be transported as object-mode outputs. Ordered mode keeps a ring buffer and waits for the next input position, which gives deterministic output at the cost of head-of-line blocking. Unordered mode reduces that wait and emits completed values promptly, so consumers must not assume source order. With `objectMode: false`, inputs are Buffers unless an upstream encoding changes them, and outputs must satisfy stream chunk rules. The one-argument `transform(worker)` form defaults concurrency to 1. If you adapt a promise, catch rejection and call the callback once; a returned promise by itself is ignored. For HTTP, database, or filesystem work, choose limits based on connection pools and service quotas, and add your own AbortController or cleanup registry if pipeline cancellation must stop active operations.

Patterns

Process objects concurrently in source ordertransform-in-order

const parallel = require('parallel-transform');

const enrich = parallel(8, (record, callback) => {
  lookupOwner(record.ownerId, (error, owner) => {
    if (error) return callback(error);
    callback(null, { ...record, owner });
  });
});

Up to eight workers can run, but a slow earlier record delays every later completed result until its position is ready.

Emit each result as soon as it finishesemit-completion-order

const transform = parallel(8, { ordered: false }, (job, callback) => {
  runJob(job, callback);
});

Unordered mode reduces head-of-line waiting, but output order is nondeterministic and must not be joined to source data by position.

Process a binary streamtransform-buffers

const transform = parallel(4, { objectMode: false }, (chunk, callback) => {
  compressChunk(chunk, callback);
});

fs.createReadStream('input.bin')
  .pipe(transform)
  .pipe(fs.createWriteStream('output.bin'));

Chunks are arbitrary Buffer boundaries, not semantic records. Independent chunk compression is invalid for formats that require one continuous codec state.

Contain transform errors with pipelinehandle-pipeline-errors

const { pipeline } = require('node:stream');

pipeline(source, transform, destination, (error) => {
  if (error) {
    console.error('Pipeline failed:', error);
    process.exitCode = 1;
  }
});

A worker error emits on the transform. Pipeline prevents an unhandled error event, but already-started worker operations continue unless your code cancels them.

Drop records during transformationfilter-stream-items

const activeOnly = parallel(4, (record, callback) => {
  if (!record.active) return callback(null, null);
  callback(null, record);
});

Both `null` and `undefined` successful results are discarded, so they cannot be emitted as meaningful object-mode values.

Adapt a promise worker to the callback APIadapt-async-function

const transform = parallel(6, (item, callback) => {
  Promise.resolve()
    .then(() => enrich(item))
    .then((value) => callback(null, value), callback);
});

Returning `enrich(item)` directly does nothing because the package does not observe promises. Ensure the promise chain reaches the callback exactly once.

Use the single-argument serial formrun-serial-transform

const normalize = parallel((row, callback) => {
  callback(null, { ...row, name: row.name.trim() });
});

When the first argument is the worker function, concurrency defaults to 1 and object mode still defaults to true.

Set concurrency and buffering explicitlybound-stream-buffer

const transform = parallel(4, {
  objectMode: true,
  highWaterMark: 8,
}, processRecord);

The package otherwise raises highWaterMark to at least the concurrency or 16. Buffering and active-worker limits are related but separate.

Reject invalid concurrency before constructionvalidate-concurrency

const concurrency = Number(process.env.WORKERS ?? 4);
if (!Number.isInteger(concurrency) || concurrency < 1) {
  throw new TypeError('WORKERS must be a positive integer');
}

const transform = parallel(concurrency, worker);

The constructor does not validate `maxParallel`; zero or invalid values can leave writes waiting forever.

Add application-owned request cancellationcancel-active-requests

const active = new Set();
const transform = parallel(5, asyncWorker);

function asyncWorker(item, callback) {
  const controller = new AbortController();
  active.add(controller);
  fetch(item.url, { signal: controller.signal })
    .then((response) => response.text())
    .then((value) => callback(null, value), callback)
    .finally(() => active.delete(controller));
}

transform.on('close', () => {
  for (const controller of active) controller.abort();
});

The package does not cancel workers itself. Own every abort handle, and make sure abort rejection reaches the callback no more than once.

Alternatives

Package	Registry	Pick it when
through2-concurrent	npm	You want a similar callback-style concurrent transform with a through2-shaped API
concurrent-transform	npm	You want another small object-stream transform that bounds concurrent asynchronous handlers
p-map	npm	Your inputs fit an iterable and you want promise-native concurrency, async functions, and clearer rejection behavior