@metreeca/pipe

A lightweight TypeScript library for composable async iterable processing.

@metreeca/pipe provides an idiomatic, easy-to-use functional API for working with async iterables through pipes, tasks, and sinks. The composable design enables building complex data processing pipelines with full type safety and minimal boilerplate. Key features include:

• Focused API › Small set of operators covering common async iterable use cases
• Natural syntax › Readable pipeline composition: items(data)(filter())(map())(toArray())
• Minimal boilerplate › Automatic undefined filtering and seamless type inference across pipeline stages
• Task/Sink pattern › Clear separation between transformations and terminal operations
• Parallel processing › Built-in { parallel: true } option for concurrent execution
• Extensible design › Easy creation of custom feeds, tasks, and sinks

Installation

npm install @metreeca/pipe
Copy

Usage

Core Concepts

@metreeca/pipe provides four main abstractions:

• Pipes : Fluent interface for composing async stream operations
• Feeds : Factory functions that create new pipes from various input sources
• Tasks : Intermediate operations that transform, filter, or process stream items
• Sinks : Terminal operations that consume streams and produce final results

Creating Feeds

Create feeds from various data sources.

import { range, items, chain, merge, iterate } from '@metreeca/pipe/feeds';

items(42);                    // from single values
items(1, 2, 3, 4, 5);         // from multiple scalar values
items([1, 2, 3, 4, 5]);       // from arrays
items(new Set([1, 2, 3]));    // from iterables
items(asyncGenerator());      // from async iterables
items(pipe);                  // from pipes
range(10, 0);                 // from numeric ranges

iterate(() => Math.random()); // from repeated generator calls

chain(                        // sequential consumption
	items([1, 2, 3]),
	items([4, 5, 6])
);

merge(                        // concurrent consumption
	items([1, 2, 3]),
	items([4, 5, 6])
);
Copy

Transforming Data

Chain tasks to transform, filter, and process items.

import { items } from '@metreeca/pipe/feeds';
import { map, filter, take, distinct, batch } from '@metreeca/pipe/tasks';
import { toArray } from '@metreeca/pipe/sinks';
import { pipe } from '@metreeca/pipe';

await pipe(
	(items([1, 2, 3, 4, 5]))
	(filter(x => x%2 === 0))
	(map(x => x*2))
	(take(2))
	(toArray())
);  // [4, 8]

await pipe(
	(items([1, 2, 2, 3, 1]))
	(distinct())
	(toArray())
);  // [1, 2, 3]

await pipe(
	(items([1, 2, 3, 4, 5]))
	(batch(2))
	(toArray())
);  // [[1, 2], [3, 4], [5]]
Copy

Parallel Processing

Process items concurrently with the parallel option in map() and flatMap() tasks.

import { items } from '@metreeca/pipe/feeds';
import { flatMap, map } from '@metreeca/pipe/tasks';
import { toArray } from '@metreeca/pipe/sinks';
import { pipe } from '@metreeca/pipe';

await pipe( // mapping with auto-detected concurrency (CPU cores)
	(items([1, 2, 3]))
	(map(async x => x*2, { parallel: true }))
	(toArray())
);

await pipe( // mapping with unbounded concurrency (I/O-heavy tasks)
	(items(urls))
	(map(async url => fetch(url), { parallel: 0 }))
	(toArray())
);

await pipe( // flat-mapping with explicit limit
	(items([1, 2, 3]))
	(flatMap(async x => [x, x*2], { parallel: 2 }))
  (toArray())
);
Copy

Consuming Data

Apply sinks as terminal operations that consume pipes and return promises with final results.

import { items } from '@metreeca/pipe/feeds';
import { some, find, reduce, toArray, forEach } from '@metreeca/pipe/sinks';
import { pipe } from '@metreeca/pipe';

await pipe(
	(items([1, 2, 3]))
	(some(x => x > 2))
);  // true

await pipe(
	(items([1, 2, 3, 4]))
	(find(x => x > 2))
);  // 3

await pipe(
	(items([1, 2, 3, 4]))
	(reduce((a, x) => a+x, 0))
);  // 10

await pipe(
	(items([1, 2, 3]))
	(toArray())
);  // [1, 2, 3]

await pipe(
	(items([1, 2, 3]))
	(forEach(x => console.log(x)))
);  // 3
Copy

Alternatively, call pipe() without a sink to get the underlying async iterable for manual iteration.

import { items } from '@metreeca/pipe/feeds';
import { filter } from '@metreeca/pipe/tasks';
import { pipe } from '@metreeca/pipe';

const iterable = pipe(
	items([1, 2, 3])(filter(x => x > 1))
);  // AsyncIterable<number>

for await (const value of iterable) {
	console.log(value);  // 2, 3
}
Copy

Working with Infinite Feeds

Use iterate() to create infinite feeds from generator functions. Tasks and sinks handle infinite feeds gracefully, processing values lazily until a limiting operator (like take()) or terminal sink stops consumption.

import { iterate } from '@metreeca/pipe/feeds';
import { filter, take } from '@metreeca/pipe/tasks';
import { forEach } from '@metreeca/pipe/sinks';
import { pipe } from '@metreeca/pipe';

await pipe(
	(iterate(() => Math.random()))
	(filter(v => v > 0.5))
	(take(3))
	(forEach(console.info))
);
Copy

Creating Custom Tasks

Tasks are functions that transform async iterables. Create custom tasks by returning an async generator function.

import { items } from '@metreeca/pipe/feeds';
import { toArray } from '@metreeca/pipe/sinks';
import type { Task } from '@metreeca/pipe';

function double<V extends number>(): Task<V, V> {
	return async function* (source) {
		for await (const item of source) { yield item*2 as V; }
	};
}

await items([1, 2, 3])(double())(toArray());  // [2, 4, 6]
Copy

Creating Custom Feeds

Feeds are functions that create new pipes.

import { items } from '@metreeca/pipe/feeds';
import { toArray } from '@metreeca/pipe/sinks';
import type { Pipe } from '@metreeca/pipe';

function repeat<V>(value: V, count: number): Pipe<V> {
	return items(async function* () {
		for (let i = 0; i < count; i++) { yield value; }
	}());
}

await repeat(42, 3)(toArray());  // [42, 42, 42]
Copy

Support

• open an issue to report a problem or to suggest a new feature
• start a discussion to ask a how-to question or to share an idea

License

This project is licensed under the Apache 2.0 License – see LICENSE file for details.