stream-pipeline-runner

Run Node.js stream pipelines with retries, progress tracking, timeouts and helpful transform wrappers.

Features

• Promise-based runPipeline wrapper around stream/promises.
• Per-transform retry policy.
• Global timeout / AbortSignal support.
• Progress callbacks (bytes / records).
• parallelTransform helper to process items with concurrency.

Quick example

import fs from 'fs';
import { runPipeline, parallelTransform } from 'stream-pipeline-runner';


await runPipeline({
source: fs.createReadStream('input.ndjson'),
transforms: [
parallelTransform(async (chunk) => {
// parse NDJSON line -> object
const obj = JSON.parse(chunk.toString());
// enrich or call API
return JSON.stringify(obj) + '\n';
}, { concurrency: 4 })
],
destination: fs.createWriteStream('out.ndjson'),
options: { timeoutMs: 30_000, onProgress: (p) => console.log(p) }
});