@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: pipe(items(data)(filter())(map())(toArray()))
• Type safety › Seamless type inference across pipeline stages and automatic undefined filtering
• Task/Sink pattern › Clear separation between transformations and terminal operations
• Parallel processing › Built-in support for concurrent task execution
• Extensible design › Easy creation of custom feeds, tasks, and sinks

Installation

npm install @metreeca/pipe

Warning

TypeScript consumers must use "moduleResolution": "nodenext"/"node16"/"bundler" in tsconfig.json. The legacy "node" resolver is not supported.

Core Usage

Note

This section introduces essential concepts and common patterns: see the API reference for complete coverage.

@metreeca/pipe provides four main abstractions:

• Pipes : Functional 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])
);

Transforming Data

Chain tasks to transform, filter, and process items. The @metreeca/core comparators and predicates modules provide helper functions for assembling complex sorting and filtering criteria.

import { pipe } from "@metreeca/pipe";
import { items } from "@metreeca/pipe/feeds";
import { toArray } from "@metreeca/pipe/sinks";
import { by } from "@metreeca/core/comparators";
import { batch, distinct, filter, map, sort, take } from "@metreeca/pipe/tasks";

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([3, 1, 4, 1, 5]))
	(sort())
	(toArray())
);  // [1, 1, 3, 4, 5]

await pipe(
	(items([{ name: "Alice", age: 30 }, { name: "Bob", age: 25 }]))
  (sort(by(x => x.age)))
	(toArray())
);  // [{ name: "Bob", age: 25 }, { name: "Alice", age: 30 }]

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

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

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
}

Advanced Usage

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())
);

Manage parallel execution flow with utilities from the @metreeca/core async module, such as throttling to control execution rate:

import { Throttle } from "@metreeca/core/async";
import { pipe } from "@metreeca/pipe";
import { items } from "@metreeca/pipe/feeds";
import { forEach } from "@metreeca/pipe/sinks";
import { map } from "@metreeca/pipe/tasks";

const throttle = Throttle({ minimum: 1000 });  // limit to max 1 request per second

await pipe(
	(items([1, 2, 3, 4, 5]))
	(map(throttle.queue))  // inject delays to enforce rate limit
		(map(async x => fetch(`/api/items/${x}`)))
		(forEach(console.log))
);

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))
);

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]

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]

Caution

When creating custom feeds, always wrap async generators, async generator functions, or AsyncIterable<T> objects with items() to ensure undefined filtering and proper pipe interface integration.

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.