@metreeca/flow

Composable primitives for asynchronous data flows.

@metreeca/flow replaces the boilerplate of async data processing with declarative data flows, read as one unit from source to result:

• Focused API: small operator set covering the common async iterable operations and composing with a natural syntax
• Type Safety: item types carried from the data source through every stage to the final result, never dictated by the functions handed to the steps
• Concurrency Control: any task run concurrently over the feed, under an explicit bound on the items in flight
• Extensible Design: custom steps written against the same contracts as the built-in ones, composing in any order

Installation

npm install @metreeca/flow
Copy

Core Usage

@metreeca/flow builds on three composable abstractions:

• Feeds: carry the items and accept the steps advancing the pipe, opened from values and data sources
• Tasks: intermediate operations transforming, filtering or reshaping the items
• Sinks: terminal operations consuming the items and computing the final result

A pipe composes a feed, any number of tasks and an optional sink. Closed by a sink, it resolves to the final result; left open, it hands back its final feed, iterated with for await to draw the items it carries.

A feed ends when the source it draws from stops yielding items: an array or a generator ends it on its own, under the iteration protocols, while a source without a natural end is bounded by the feed it is opened with, or downstream: see Bounding Feeds.

The three combine into a single pipe. This example draws a set of URLs, retrieves them concurrently, parses each response as JSON and collects the documents:

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

const urls = [1, 2, 3, 4, 5, 6].map(page => `https://example.com/docs?page=${page}`);

const docs = await pipe(
	(items(urls))
	(fork(4, map(url => fetch(url))))  // never more than 4 requests running at a time
	(map(response => response.json())) // JSON responses
	(toArray())                        // the parsed documents
);
Copy

The contracts are the whole of the API: the built-in tasks and sinks are ordinary functions honouring them, with no privileged core underneath. Custom steps honour the same contracts, so they compose with the built-in ones in any order, items() turning any value, source or generator into a conforming feed. Every step is handed a feed in turn, so a custom one either draws the items itself or passes the feed on to steps already available.

Creating Feeds

Start a pipe with a feed, adapting values and data sources into a feed ready for composition.

Generators

Open a feed over values, ranges or an external source drawn one item at a time.

import { done, inlet, items, range } from '@metreeca/flow/feeds';

items([1, 2, 3, 4, 5]);                         // from arrays
items(new Set([1, 2, 3]));                      // from iterables
items(asyncGenerator());                        // from async iterables, an existing feed among them
items(42);                                      // from single values
items(fetch(url).then(r => r.json()));          // from promises, awaited on consumption; the payload enters whole

range(10, 0);                                   // from numeric ranges

inlet(() => Math.random());                     // from an external source, endless unless the source ends it
inlet(() => queue.size ? queue.poll() : done);  // ended by the source reporting the done marker
Copy

items() contributes a source according to its shape: iterables and async iterables contribute the items they yield, while any other value, a promise once awaited, is contributed whole as a single item. The source is drawn exactly as handed over, so a feed opened from a generator object runs dry after the first pass, while one opened from a repeatable source, an array or a set among them, is consumed afresh at each. A feed reported by a task is drained by a single pass whatever it draws from: consume a composition twice by building it afresh from the feed it opens with.

Several feeds are combined into one by carrying them in a feed of their own and splicing it with flat() or join(), in a pipe or in place as flat(items([a, b])), or by naming them at the call site of mix(): see Splicers.

Transforming Data

Chain tasks to reshape the feed and map its items.

Operators

Select, reorder and inspect items, leaving their type unchanged but for the undefined dropped by trim().

import { distinct, filter, map, peek, skip, sort, take, trim } from '@metreeca/flow/tasks';

await pipe(
	(items([1, 2, 3, 4, 5]))
	(filter(n => n%2 === 0))
	(toArray())
);  // [2, 4]

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

await pipe(
	(items([1, 2, 3, 4]))
	(trim(map(n => n%2 ? n*10 : undefined)))
	(toArray())
);  // [10, 30], as items mapped to undefined are dropped

await pipe(
	(trim(items([undefined, 1, undefined, 2])))
	(toArray())
);  // [1, 2], as the feed is trimmed in place

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

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

await pipe(
	(items([1, 2, 3]))
	(peek(n => console.log(n)))
	(toArray())
);  // logs 1, 2, 3; [1, 2, 3]
Copy

Transformers

Map items to values of a different type, either one by one, in groups, or over the feed as a whole.

import { batch, chain, group, map } from '@metreeca/flow/tasks';

await pipe(
	(items([1, 2, 3]))
	(map(n => n*2))
	(toArray())
);  // [2, 4, 6]

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

await pipe(
	(items([1, 2, 3, 4, 5]))
	(group(n => n%2))
	(toArray())
);  // [[1, [1, 3, 5]], [0, [2, 4]]]

await pipe(
	(items([1, 2, 3, 4]))
	(chain(async feed => (await feed(toArray())).slice(-2)))
	(toArray())
);  // [3, 4], as the last items are known only once the feed runs dry
Copy

chain() hands the feed to a mapper and carries on with the items it computes, so a step that cannot decide before the feed runs dry (reconciling it against a stored snapshot, ranking it, clearing it against a quota) is written as an ordinary asynchronous function of the feed rather than as a generator. Sinks already available are lifted back into the pipe the same way, as chain(toSet()) is to carry on with the distinct items alone. The items are supplied either as a batch listing them or as a feed yielding them, handed back as they are or awaited, so a step composing the feed it draws from with tasks already available is spared the generator as well. Whatever the mapper computes is emitted item by item, so the items carried on need be neither the ones drawn, nor as many, nor of the same type.

Splicers

Splice several feeds into the pipe: the nested feeds a source carries or a task opens, in source order or interleaved as their items become available, the concurrent runs of a task, the branches every item is handed to, the two tasks a condition chooses between for each item, or sources named at the call site of the task splicing them in.

import { filter, flat, fork, join, map, mix, take, tee, when } from '@metreeca/flow/tasks';

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

await pipe(
	(flat(items([items([1, 2]), items([3, 4])])))
	(toArray())
);  // [1, 2, 3, 4], as the feed is spliced in place

await pipe(
	(items([1, 2, 3]))
	(flat(map(n => [n, n*10])))
	(toArray())
);  // [1, 10, 2, 20, 3, 30], as each item is expanded into the items it is mapped to

await pipe(
	(items([items([1, 2, 3]), items([4, 5, 6])]))
	(flat(map(feed => feed(take(2)))))
	(toArray())
);  // [1, 2, 4, 5], as the quota is scoped to each nested feed

await pipe(
	(items([slow, fast]))  // slow yields 1, 2; fast yields 3, 4
	(join())
	(toArray())
);  // [3, 4, 1, 2], as the faster feed reports first

await pipe(
	(items(ids))
	(fork(4, retrieve()))
	(toArray())
);  // at most 4 items in flight, results in completion order

await pipe(
	(items([1, 2, 3]))
	(tee(map(n => n*2), filter(n => n > 2)))
	(toArray())
);  // 2, 4, 6 from the doubling branch and 3 from the filtering one, interleaved in no defined order

await pipe(
	(items([1, 2, 3, 4]))
	(when(n => n%2 === 0, map(n => n*10), map(n => -n)))
	(toArray())
);  // 20, 40 from the even route and -1, -3 from the odd one, interleaved in no defined order

await pipe(
	(items([1, 2]))
	(mix(items([3, 4]), 5))
	(toArray())
);  // 1, 2 from the feed, 3, 4 from the mixed feed and 5 as a single item, in no defined order
Copy

flat() splices one level only: a feed carried by a nested feed is reported as an item, ready for a further splice. Nested feeds may be any sync or async iterable, so an array is spliced as is, with no feed opened around it. The optional task opens the feeds to splice, drawing from the whole feed, so an item mapped to an array or to a feed of its own is expanded in place; scope a task to each nested feed by applying it within map(), where the source already carries feeds. Handed a feed of feeds instead of a task, flat() splices it in place, with no pipe routing it. join() splices nested feeds the same way, in a pipe or in place, but opens every one as soon as it is reported and emits items as they become available, so output order is not preserved and nothing bounds the number of feeds open at once.

tee() fans out instead of splitting: every branch is applied to the whole feed and handed every item, so a stateful branch decides on every item, unlike a forked run; the items the branches report are interleaved as join() interleaves nested feeds. Branches draw in lockstep, so nothing is held beyond the item on offer and the source advances at the pace of the slowest branch: pacing and long-running work belong downstream of the fan-out, while a branch closing early, as take() does, or running dry on its own drops out and stops holding back the others.

when() routes instead of fanning out: each item is tested once and handed to one of two tasks, the first if it meets the condition and the second otherwise. Each task decides only on the items routed to it, so a quota or a deduplication covers its own share of the feed. The two tasks draw in lockstep and interleave their items as tee() branches do.

mix() mixes in sources of its own, rather than splicing the nested feeds a source carries, so a pipe gathering items from several origins names them at the call site and keeps the fan-in to a single stage. Each source is opened as items() opens it, and none of them draws from the feed or paces the others: one idling or long-running delays its own items alone.

See Concurrent Processing for the bounds fork() sets and the state it tolerates.

Consuming Data

Close a pipe with a sink, consuming the items and computing the final result.

Predicates

Test the feed against a condition, stopping as soon as the outcome is decided.

import { every, some } from '@metreeca/flow/sinks';

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

await pipe(
	(items([2, 4, 6]))
	(every(n => n%2 === 0))
);  // true
Copy

Aggregators

Reduce the feed to a single value, resolving to undefined on an empty feed where no result is defined. sum() and avg() handle number and bigint feeds alike, rounding bigint means to the nearest integer, halves away from zero. min() and max() rank items with the same comparators as sort(), resolving to the first of equally ranking ones.

import { avg, count, max, min, sum } from '@metreeca/flow/sinks';

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

await pipe(
	(items([1, 2, 3, 4, 5]))
	(sum())
);  // 15

await pipe(
	(items([1, 2, 3, 4]))
	(avg())
);  // 2.5

await pipe(
	(items([1n, 2n, 4n]))
	(avg())
);  // 2n

await pipe(
	(items([3, 1, 4, 1, 5]))
	(min())
);  // 1

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

Scanners

Retrieve a single item from the feed or fold it into a value of an arbitrary type. Both find() and seek() retrieve the first item of the feed, so the first item meeting a condition is retrieved by drawing from a filter() upstream. Where the feed carries no item, find() resolves to undefined, leaving the choice of a fallback to the caller, while seek() fails, so the item it hands back is usable as is.

import { find, reduce, seek } from '@metreeca/flow/sinks';
import { filter } from '@metreeca/flow/tasks';

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

await pipe(
	(items([1, 2, 3, 4]))
	(filter(n => n > 10))
	(seek())
);  // fails, as the filtered feed carries no item

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

Collectors

Collect items into a container. The result is deeply immutable, freezing both the container and the items, keys and values collected into it.

import { toArray, toMap, toObject, toSet, toString } from '@metreeca/flow/sinks';

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

await pipe(
	(items([1, 2, 2, 3]))
	(toSet())
);  // Set(3) { 1, 2, 3 }

await pipe(
	(items([{ id: 1, name: "Alice" }, { id: 2, name: "Bob" }]))
	(toMap(x => x.id, x => x.name))
);  // Map(2) { 1 => "Alice", 2 => "Bob" }

await pipe(
	(items([{ id: 1, name: "Alice" }, { id: 2, name: "Bob" }]))
	(toObject(x => x.id, x => x.name))
);  // { 1: "Alice", 2: "Bob" }

await pipe(
	(items([1, 2, 3]))
	(toString(" - "))
);  // "1 - 2 - 3"
Copy

Effects

Consume the feed for its side effects, resolving to the number of items processed.

import { each } from '@metreeca/flow/sinks';

await pipe(
	(items([1, 2, 3]))
	(each(n => console.log(n)))
);  // logs 1, 2, 3; 3
Copy

Advanced Usage

Manual Iteration

Feeds are async iterables, so a pipe left open without a sink is iterated directly, pulling items as the surrounding code is ready for them. Bracketing the composition in pipe() hands back the feed it ends with, reading as one unit exactly as a closed pipe does.

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

const evens = pipe(
	(items([1, 2, 3, 4]))
	(filter(n => n%2 === 0))
);  // Feed<number>

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

Bounding Feeds

Use inlet() to open infinite feeds over an external source. Items are pulled lazily, so an infinite feed is consumed only as far as the pipe demands: bound it with a task like take(), or with a sink deciding its outcome early, such as some(), every(), find() or seek().

A source knowing when it is exhausted ends the feed itself, either reporting the done marker in place of a value or wrapped in a generator of its own; an AbortSignal bounds it from the outside instead, ending it as an exhausted source does and leaving the items already contributed to the pipe that consumes them.

inlet(() => queue.size ? queue.poll() : done);           // every value queued, until the queue runs dry

inlet(() => cursor.next(), AbortSignal.timeout(1_000));  // every value reported within the deadline
Copy

Two hazards follow, the first on infinite feeds alone, the second on any feed large enough:

• never completing: sort(), group(), an unbounded batch() and a chain() whose mapper draws the feed entire emit nothing before the whole feed is drawn, and every sink but some(), every(), find() and seek() needs every item
• exhausting memory: the first three of those tasks materialise the feed whole and a chain() holds whatever its mapper retains and computes, distinct() retains every key seen, join() holds a pending item per open nested feed and an uncapped fork() a run per item drawn, and the collectors build the whole container before resolving

Batch by a positive size, cap the runs of a fork, or bound the feed upstream.

The API reference classifies every feed, task and sink along four axes:

• bounded or infinite, for how far a feed goes
• incremental or exhaustive, for how much of a feed a task or sink draws before emitting or resolving
• streaming or materialising, for what is held in memory
• stateless or stateful, for whether the outcome depends on the items drawn before it

import { inlet } from '@metreeca/flow/feeds';
import { filter, take } from '@metreeca/flow/tasks';
import { each } from '@metreeca/flow/sinks';
import { pipe } from '@metreeca/flow';

await pipe(
	(inlet(() => Math.random()))
	(filter(n => n > 0.5))
	(take(3))
	(each(n => console.info(n)))
);
Copy

Concurrent Processing

Wrap any task with fork() to interleave several runs of it over the same feed: the runs draw from the same source, each item going to exactly one of them, and results are emitted as soon as they are ready rather than in source order.

Runs are interleaved on the event loop rather than executed in parallel: the bound covers the asynchronous operations overlapping at any time, not the computation carried out simultaneously, so forking a task that blocks the event loop buys nothing.

The number of runs bounds how far the task reads ahead of the downstream consumer, so it controls memory usage and backpressure, not the rate at which work is submitted. A capped fork starts every run when the feed is opened, so the task is invoked exactly as many times as the bound, whether or not the source is fast enough to keep every run busy.

Pass 0 to leave the fork uncapped, opening a run for each item the source delivers: throughput is limited only by the source, at the cost of an unbounded number of items in flight. The number of runs must be an integer, otherwise a TypeError is thrown; negative values are treated as 1, that is, as sequential processing.

A forked task never sees the whole feed. The task is a single function invoked once per run: state it initialises on invocation, as distinct(), sort(), take(), skip(), batch(), group() and chain() do, is tracked per run rather than across the feed as a whole, while state captured in its enclosing closure is shared by every run and accessed concurrently. Fork a stateful task only where its outcome is sound on the items one run happens to draw.

Where the state belongs to one item rather than to the feed, open a pipe per item and collapse the pipes with join(map(…)), which keeps it scoped to that item while still drawing from every pipe at once. fork() composes within each nested feed in turn, as join(map(feed => feed(fork(…)))), bounding the work carried out inside it.

Control the rate at which work is submitted with utilities from the async module of @metreeca/core, such as throttling:

import { createThrottle } from '@metreeca/core/async';
import { pipe } from '@metreeca/flow';
import { items } from '@metreeca/flow/feeds';
import { each } from '@metreeca/flow/sinks';
import { fork } from '@metreeca/flow/tasks';

const throttle = createThrottle({ minimum: 1000 });  // at most 1 request per second

await pipe(
	(items(ids))
	(fork(4, retrieve(throttle)))  // inject delays to enforce the rate limit
	(each(x => console.log(x)))
);
Copy

Creating Custom Feeds

A custom feed opens a pipe over a source of your own.

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

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

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

Creating Custom Tasks

A custom task extends a pipe, consuming the items of a feed and reporting a new one. The transformation is most easily written as an async generator handed to items(), with items to be dropped left unyielded.

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

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

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

Creating Custom Sinks

A custom sink closes a pipe, consuming the items and returning a promise for the final result. A computation delegating to operations already available applies them to the feed it draws from, which is drained by a single pass, however repeatable the source behind it.

import { pipe } from '@metreeca/flow';
import { items } from '@metreeca/flow/feeds';
import type { Sink } from '@metreeca/flow';

function histogram<V>(): Sink<V, Map<V, number>> {
	return async source => {

		const counts = new Map<V, number>();

		for await (const item of source) { counts.set(item, (counts.get(item) ?? 0)+1); }

		return counts;
	};
}

await pipe(
	(items(["a", "b", "a"]))
	(histogram())
);  // Map(2) { "a" => 2, "b" => 1 }
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.