pipelino.
GitHub

Give your async work a sense of flow.

A staged pipeline for async work, in the browser and in Node. You write the stages; pipelino moves every item through them.

Get started

Watch a pipeline run

24 photos move through three stages. The scheduler is the real library; only the work is simulated.

Upload waits until every thumbnail exists.

24 photosReady

Inspect

concurrency: 4
Ready0 / 24

Thumbnail

concurrency: 3
Ready0 / 24

Upload

barrier
batch: 6, concurrency: 2
Ready0 / 24
0 uploaded0%
QueuedRunningDone
pipeline.push(photos)Ready0.0s

Quick start

One object describes the whole flow. Push items and listen for progress.

upload.ts 
import { createPipeline } from 'pipelino';

const pipeline = createPipeline<Photo>({
  stages: [
    { name: 'inspect', concurrency: 4, run: inspect },
    { name: 'thumbnail', concurrency: 3, run: thumbnail },
    {
      name: 'upload',
      barrier: true,
      batch: { size: 6 },
      runBatch: (photos, ctx) => upload(photos, ctx.signal),
    },
  ],
});

pipeline.on('progress', (p) => render(p.overall.percent));
pipeline.on('done', ({ failed }) => report(failed));
pipeline.push(photos);

  1. Stages, not queues

    Items move from stage to stage on their own. You never hand them from one queue to the next.

  2. Results merge into the item

    Return an object from a stage and its fields are assigned to the item. There is no second data structure to keep in sync.

  3. Streaming by default

    An item enters the next stage the moment it leaves the previous one. barrier: true holds a stage until every item has caught up.

Features

The parts of a multi-step flow you would otherwise write by hand.

Barrier stages

Hold a stage until no item is still behind it: upload only once every thumbnail exists.

barrier: true

Batching by count and weight

Group ready items by count and by a weight such as bytes. A batch never waits to fill up.

batch: { size, weight, weigh }

Pause, resume, cancel

Pausing lets running work finish. Cancelling aborts the AbortSignal every stage receives.

pause() · resume() · cancel()

Live progress

Done, failed and running counts per stage and overall, kept current without scanning the items.

on('progress', render)

Isolated failures

A stage that throws fails only that item. It skips the remaining stages; the others keep moving.

on('stage:error', handle)

Changes while it runs

Push more items, enter them mid-pipeline, skip a stage per item, or append stages later.

push(items, { startAt }) · addStages()

Compared with queues

A queue limits concurrency for one kind of task. pipelino adds the stages, batching and bookkeeping around it.

Capability pipelino p-queue fastq async.queue p-limit RxJS
Several stages in one object ✓ – – – – operators
Barrier stages ✓ – – – – manual
Batching by count ✓ – – cargo – bufferCount
Batching by weight (bytes) ✓ – – – – manual
Concurrency per stage ✓ per queue per queue per queue per limit mergeMap
Pause and resume ✓ ✓ ✓ ✓ – manual
Cancel with an AbortSignal ✓ clear, own signal kill kill – unsubscribe
Skip, enter mid-flow, add stages ✓ – – – – –
Results merged into items ✓ promise per task callback callback promise per task stream
Progress per stage and overall ✓ counters counters counters counters manual
Runtime dependencies 0 2 1 0 1 1
Minified and gzipped 2.6 kB 4.4 kB 1.2 kB 8.1 kB 0.9 kB ~18 kB

p-queue 9.3.3, fastq 1.20.3, async 3.2.6 (whole library), p-limit 7.3.2. Sizes are esbuild bundles with dependencies, gzip -9; RxJS is an estimate for a typical set of operators. fastq and p-limit are smaller because they do less.

60,000 stage runs in 10 ms

20,000 items through three no-op stages. The scheduler is never the bottleneck.

Method and full results
Time to move 20,000 items through 3 stages, concurrency 8
pipelino10 ms
async.queue, one per stage16 ms
fastq, one per stage19 ms
p-limit, one per stage30 ms
p-queue, one per stage39 ms

Node 22, Apple M3 Max, median of 7 runs, September 2026. Recorded results, not measured live.

API

One function, one object. Everything is typed over your item type.

createPipeline({ stages })
A stage has a name, run or batch with runBatch, and optionally concurrency, barrier and skip.
push(items, { startAt? })
Queues items and returns how many were added. Ids already queued or running are ignored.
remove(ids)
Drops queued, running or finished items.
addStages(stages)
Appends stages. Finished items re-enter at the first new one.
pause() and resume()
Stop and continue scheduling. Running work finishes.
cancel() and reset()
Abort the signal and drop unfinished items. reset() also forgets the finished ones.
progress()
Done, failed, running, total and percent, overall and per stage.
items()
A snapshot per item: its state, current stage and error.
on(event, listener)
stage:done, stage:error, item:done, progress, status and done. Returns the unsubscribe function.
status
'idle', 'running' or 'paused'.

Full reference

Questions

Where does it run?

In current browsers, Node 18 and later, Deno and Bun. It ships as ES2022 modules and has no dependencies.

What happens when a stage throws?

That item is marked failed, skips the remaining stages and is listed in done.failed. Other items are not affected.

Can I add items while it runs?

Yes. push() works at any time, and done fires again once the new items have settled.

Does it retry failed items?

Not on its own. Push a failed item again and it runs from the first stage, or from a later one with startAt.

Pairs with imagefast

Read dimensions and make thumbnails on a worker pool in the browser, then let pipelino batch the uploads.

Meet imagefast

Add it to your app

Read the docs