Module tasks

Intermediate operations that transform, filter or reshape the items of a feed.

Tasks apply to a Feed and yield a new feed, so they chain freely into longer pipes. Items are processed lazily, sequentially and in source order, unless a task reorders them, interleaves the nested feeds carrying them, hands each of them to several tasks at once or wraps another to run it concurrently, trading output order for throughput.

A task wrapping others either hands each of them the whole feed, as flat, join and tee do, leaving whatever state they initialise on invocation to decide on every item, or invokes one once per run over a share of the items, as fork does, scoping that state to the run rather than to the pipe as a whole. The feed a task draws from and the one it reports are both drained by a single pass, however repeatable the source behind them: a composition to be consumed twice is built afresh from the feed it opens with.

The item type of a task is taken from the feed it is applied to, or from the Task type it is declared under, never from the predicate, selector, mapper or comparator handed to it: a function accepting any item, console.log or Boolean among them, leaves the feed type untouched, while one accepting a different type is rejected. A task composed on its own, outside a pipe, states its item type in the type it is declared under or as an explicit type argument.

Every task is classified along three axes:

• incremental or exhaustive, for how much of the feed it draws before emitting
• streaming or materialising, for what it holds in memory
• stateless or stateful, for whether its outcome depends on the items drawn before it

Custom Tasks extend a pipe, drawing 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