Skip to content

reference.md

Latest commit

 

History

History
988 lines (727 loc) · 39.1 KB

reference.md

File metadata and controls

988 lines (727 loc) · 39.1 KB
Raw

Pipelean Reference Documentation

Overview

functional.js is an async programming library that provides core utilities for handling asynchronous operations, error management, data transformations, and functional composition patterns in JavaScript. The library is designed with a pragmatic philosophy: avoid heavy abstractions, use eager execution, and provide clear, predictable error handling.

Key Principles

  • Pragmatic: Plain JavaScript, eager execution, sequential processing — plus join() for concurrent fork/join over named tasks.
  • First class error handling: Multiple error strategies for different use cases

Table of Contents

Error strategies

  • failFast - Error Strategy Identifier (aliases: stopOnError, fail)
  • rethrow - Error Strategy Identifier (throws on first error)
  • failLate - Error Strategy Identifier
  • collect - Error Strategy Identifier
  • skip - Error Strategy Identifier
  • Callbacks - Error handling callbacks

Iterators

Horizontal

  • series - Stateless Sequential Execution
  • scan - Stateful Sequential Transformation
  • reduce - Pure Reduction
  • filter - Stateless Selection
  • stopWhen - Source Adapter for Predicate-Based Early Exit (alias of sorts: the while you don't need)
  • forever - Source Adapter for Unbounded Ticking (the while (true) you don't need)

Composition

  • pipe - Vertical Composition
  • flow - Stateful Accumulation Across One Input
  • join - Fork/join over Named Concurrent Tasks

Misc

  • assign - Conditional property assignment for flow
  • where - Object-pattern predicate
  • retry - Configurable Retry Logic
  • tryCatch - Single Function Lifecycle Hooks

Sync Variants

Error strategies

failFast

Purpose: Error strategy identifier that stops immediately on the first error.

Alias: stopOnError

Use when: Critical operations where failure means the entire pipeline is invalid.

Behavior:

  • Sets failure: {item, error, index} on first error
  • Calls onError({item, error, index, total}), then onFailure({item, error, index}) immediately
  • Stops iteration

Return Format: {results: [], errors: [], failure: {item, error, index}}

Example:

import {series, failFast} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2) throw new Error('Error')
  return item * 2
}, {strategy: failFast})

// result = {results: [], errors: [], failure: {item: 2, error: Error(...), index: 1}}

rethrow

Purpose: Error strategy identifier that throws the error immediately on the first failure. Does NOT return a structured result on failure.

Use when: "Let it crash" / fail-early patterns where the caller is expected to handle errors externally (e.g. via try/catch).

Behavior:

  • Throws the error on first failure
  • Does NOT call onFailure
  • Does NOT call onError in series

Return Format: On success: {results, errors: [], failure: false}

Example:

import {series, rethrow} from 'pipelean'

try {
  const result = await series([1, 2, 3], async item => {
    if (item === 2) throw new Error('Error')
    return item * 2
  }, {strategy: rethrow})
} catch (error) {
  // error = Error('Error')
}

collect

Purpose: Error strategy identifier that collects all errors and continues processing (default for series and filter).

Use when: Batch operations, logging scenarios, background tasks.

Behavior:

  • Collects all errors in errors array
  • Sets failure: false
  • Does NOT call onFailure

Return Format: {results, errors: [{item, error, index}], failure: false}

Example:

import {series, collect} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2 || item === 4) throw new Error('Error')
  return item * 2
}, {strategy: collect})

// result = {results: [2, 6], errors: [{item: 2, error: ..., index: 1}], failure: false}

failLate

Purpose: Error strategy identifier that collects all errors and returns failure: {errors} at the end.

Use when: Application-layer needs to detect if any error occurred.

Behavior:

  • Collects all errors in errors array
  • Sets failure: {errors} after loop completes (only if errors.length > 0)
  • Calls onFailure({errors}) if failure is truthy

Return Format: {results, errors: [{item, error, index}], failure: {errors}} (if any errors occurred)

Example:

import {series, failLate} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2 || item === 4) throw new Error('Error')
  return item * 2
}, {strategy: failLate})

// result = {results: [2, 6], errors: [{item: 2, error: ..., index: 1}], failure: {errors}}

skip

Purpose: Error strategy identifier that ignores errors entirely (no collection), but onError is still called if present.

Use when: Best-effort processing, some failures are acceptable.

Behavior:

  • Ignores errors (no collection, errors stays empty)
  • Sets failure: false
  • Does NOT call onFailure

Return Format: {results, errors: [], failure: false}

Example:

import {series, skip} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2) throw new Error('Error')
  return item * 2
}, {strategy: skip})

// result = {results: [2, 6], errors: [], failure: false}

Callbacks

onError

Optional callback for verification/telemetry (logging, metrics).

  • Called for every handled item error in series
  • Does NOT affect control flow
  • Receives {item, error, index, total}; total is omitted when unknown
  • Use for: logging, metrics, external error reporting
await series(items, fn, {
  strategy: skip,
  onError: ({item, error}) => console.error('Error:', item.id, error.message)
})

onFailure

Optional callback for application-layer error handling (UI updates, notifications).

  • Called when failure is truthy
  • Depends on strategy:
    • failFast: called with {item, error, index}
    • failLate: called with {errors}
    • collect / skip: NOT called (failure is false)
await series(items, fn, {
  strategy: failLate,
  onFailure: (failure) => {
    if (failure.errors) {
      showToast('Some items failed')
    } else {
      showToast(`Item ${failure.item} failed: ${failure.error.message}`)
    }
  }
})

Iterators

series

Purpose: Stateless sequential execution over an iterable. Runs a function on each item, one at a time.

Type: (items, fn, opts?) => Promise<Outcome> — immediate mode Curried: (fn, opts?) => (items) => Promise<Outcome>

Parameters:

  • items: An iterable (array, async iterable, generator)
  • fn(item, index): The function to apply. Returns the mapped value, or throws, or returns undefined to drop the item.
  • opts (optional):
    • strategy: Error strategy (default: collect). failFast, collect, failLate, skip, rethrow.
    • onProgress({item, result, index, total}): Called after each successful item. NOT called for errors or undefined drops.
    • onError({item, error, index, total}): Called for each handled item error. Does not affect control flow.
    • onFailure(failure): Called when failure is truthy. Receives {item, error, index} for failFast, {errors} for failLate.
    • onSourceError({error, index}): Called when the iteration itself throws (a source error). Never called under rethrow; mutually exclusive with onError.
    • take: Limit items processed.
    • total: Explicit planned input count for progress/error callbacks. If omitted, series uses a cheap known input size when available. If take is set, callback total is limited to Math.min(take, knownTotal). If no total is known, the total key is omitted.
    • pause: Milliseconds between successful items.
    • pauseOnErrors: Whether to also pause after errors (default: false).

Return: {results, errors, sourceErrors, failure} where failure is false on success. Collected errors are {item, error, index}. sourceErrors is an array of source-error contexts, empty when the source completed cleanly (see Source errors).

Usage Example:

import { series, failFast } from 'pipelean'

// Basic: double each number
const { results } = await series([1, 2, 3], x => x * 2)
// results = [2, 4, 6]

// With failFast: stop on first error
const result = await series(items, fn, { strategy: failFast })

// With pause for rate limiting
const result = await series(endpoints, fn, { pause: 500 })

// With UI progress
await series(items, saveItem, {
  onProgress: ({index, total}) => updateProgress(index + 1, total),
  onError: ({item, error}) => reportItemError(item.id, error),
})

// Curried: create a reusable transform
const double = series(x => x * 2)
const { results } = await double([1, 2, 3])

// Integration with pipe for filtering
import { pipe } from 'pipelean'
const { results } = await series(items, pipe(
  x => x.active ? x : undefined,  // drop inactive
  x => x.name,                     // extract name
))

Source errors

A source error is an error thrown by the iteration itself — the for await step (e.g. an async generator dying mid-stream because a DB connection was lost). This is distinct from an operation error, which is your fn throwing on an item.

A dead source is terminal: no more items can arrive, so iteration ends under every strategy. The error never reaches onError.

Every return path gains one additive field:

  • sourceErrors: array of {error, index} contexts — no item key, because a source error has no item. index is the iteration index of the failed .next() call (a generator that yields 2 items then throws produces index: 2). Empty array when the source completed cleanly.

New option:

  • onSourceError({error, index}): telemetry callback with the same contract as onError. Never called under rethrow. onError and onSourceError are mutually exclusive per error.

Per-strategy behavior when the source dies:

Strategy Behavior
collect Results preserved, context recorded in sourceErrors, failure: false
failFast failure = the source context, results cleared, accumulator (value in reduce mode) preserved
failLate Results preserved, source contexts merged into end-of-run aggregation: failure = {errors: [...errors, ...sourceErrors]}
skip Source errors ignored entirely: sourceErrors: [], results preserved, failure: false
rethrow Throws immediately; no callbacks called 
async function* pages(db) {
  while (true)
    yield await db.nextPage() // throws when the connection drops
}

const result = await series(pages(db), processPage, {
  strategy: collect,
  onSourceError: ({error, index}) => log.warn('source died', {error, index}),
})
// result.results holds every page processed before the drop
// result.sourceErrors = [{error, index}]

Cleanup follows native generator semantics: the generator's own finally runs during propagation, early exit via take still closes the iterator through iterator.return(), and if a finally itself throws, that error supersedes the original source error.

scan

Purpose: Stateful sequential transformation - transforms each item and accumulates results.

Type: (iterable, scanner, initialValue, opts?) => Promise<Outcome>

Parameters:

  • iterable: An async iterable (array, generator, or any object implementing the iteration protocol)
  • scanner: A function with signature (accumulator, item, index) => newAccumulator
  • initialValue: The starting value for the accumulator
  • opts (optional):
    • strategy: Error strategy (default: failFast). failFast, collect, failLate, skip, rethrow.
    • onError: Called for each handled item error. Does not affect control flow.
    • onFailure: Called when failure is truthy.
    • onSourceError: Called when the iterable itself throws.
    • storePartialResults: true (default) returns {results, …}; false returns {value, …} (see reduce).
    • scan does not support onProgress, pause, or take.

Return Type: A Promise that resolves to an object containing:

  • results: Array of intermediate results (or [] on failFast failure)
  • errors: Array of errors encountered (empty for failFast, skip, throw)
  • sourceErrors: Array of {error, index} contexts for errors thrown by the iteration itself; empty when the source completed cleanly
  • failure: false on success; {item, error, index} for failFast; {errors} for failLate (source errors are merged into the aggregated errors array)
  • value: Final accumulated value when storePartialResults: false. Preserved under every strategy except rethrow, even when the source dies mid-stream

Source errors: when the iterable itself throws mid-stream (e.g. an async generator dying), the accumulator and partial results collected so far are preserved per strategy and the context is reported through onSourceError instead of onError. See Source errors under series for the full per-strategy behavior.

Key Characteristics:

  • Stateful: Each transformation depends on the previous result
  • Accumulates: Both successful results and errors for inspection
  • Index Tracking: Provides index of each item for correlation

Usage Example:

import { scan } from './functional.js'

// Track insertions in a database
const { results, errors } = await scan(
  records,
  async (acc, record) => {
    const inserted = await db.insert(record)
    return acc + inserted  // Accumulate count
  },
  0  // Initial count
)

reduce

Purpose: Pure reduction - transforms each item sequentially and returns only the final accumulated value. A convenience wrapper around scan with storePartialResults: false baked in.

Type: (iterable, scanner, initialValue, opts?) => Promise<Outcome>

Parameters:

  • iterable: An async iterable (array, generator, or any object implementing the iteration protocol)
  • scanner: A function with signature (accumulator, item, index) => newAccumulator
  • initialValue: The starting value for the accumulator
  • opts (optional):
    • strategy: Error strategy (default: failFast). failFast, collect, failLate, skip, rethrow.
    • onError: Called for each handled item error. Does not affect control flow.
    • onFailure: Called when failure is truthy.
    • onSourceError: Called when the iterable itself throws.
    • reduce does not support onProgress, pause, or take.

Return Type: A Promise that resolves to an object containing:

  • value: The final accumulated value. Preserved under every strategy except rethrow, even when the source dies mid-stream
  • errors: Array of errors encountered (empty for failFast, skip, throw)
  • sourceErrors: Array of {error, index} contexts for errors thrown by the iteration itself; empty when the source completed cleanly
  • failure: false on success; {item, error, index} for failFast; {errors} for failLate

Key Characteristics:

  • Single value: Returns {value, errors, failure} instead of intermediate results
  • No boilerplate: No need for results.at(-1) || 0 — the final value is direct
  • Error handling: Same error strategies as scan and series

Usage Example:

import { reduce } from 'pipelean'

// Sum track durations into album total
const {value: totalDuration} = await reduce(
  tracks,
  (accumulator, {duration}) => accumulator + duration,
  0,
)
// totalDuration = 22 (no intermediate array, no .at(-1))

When to use:

  • Summing, counting, concatenating
  • Any reduction where you only need the final result
  • When scan's intermediate results array is unnecessary overhead

When to use scan instead:

  • When you need every intermediate accumulator value (e.g. dependent sequential operations where each step's output is meaningful)

Alias: scanReduce is kept as an alias of reduce for backward compatibility and points to the same function.

filter

Purpose: Stateless selection tool - filters items from an iterable based on a predicate function. Delegates to series internally: the predicate is converted to a transform that returns the original item (keep) or undefined (drop).

Type: (...args) => Promise<Outcome> | filterFunction

Parameters:

  • predicate: A function (item, index) => truthy | falsy, or a plain object pattern (converted via where())
  • items: The iterable to filter (immediate mode)
  • opts (optional): Options passed through to series

Options: Same as series — strategy, onError, onFailure, take, total, onProgress, pause, pauseOnErrors.

Return Type: { results, errors, failure } — same shape as series:

  • results: Original items where the predicate returned truthy
  • failure: false on success (no errors), {item, error, index} for failFast, {errors} for failLate

Key Characteristics:

  • The predicate's return value is never placed into results — only truthiness is checked, and the original item is what gets kept or dropped.
  • Pattern objects are supported: filter(users, {active: true}) works via where(). Anything implementing the iteration protocols (arrays, generators, source adapters like stopWhen) is always treated as a source, never as a pattern.

Usage Example:

import { filter } from 'pipelean'

const adults = await filter(
  users,
  user => user.age >= 18,
)
// result.results = [user1, user3, ...] — original items, not predicate output

stopWhen

Purpose: Source adapter that stops pulling from an iterable as soon as a predicate is truthy. The standard way to cancel or predicate-stop any source consumed by series, scan, reduce, filter, or a raw for await.

Type: (items, predicate?) => asyncIterable Sync: (items, predicate?) => iterable

Parameters:

  • items: An array or (async) iterable — generators and paging sources included
  • predicate(item, index): Checked before each item is yielded. Truthy → stop. Defaults to () => false (never stops). Sync only.

Key Characteristics:

  • Check before yield: the triggering item is pulled from the source but never offered downstream — "cancel before work"
  • Pull-lazy: once stopped, the underlying source is abandoned mid-stream; native cleanup (iterator.return(), generator finally) still runs
  • Clean completion, not an error: consumers see a finished source — failure: false, empty sourceErrors. Cancel is a shorter run, not a source death
  • Composable: zero changes in series / scan / reduce / filter; they just see an iterable that ends early. Works with raw for await too
  • Length forwarding: when the source has a numeric length (arrays), it is forwarded on the wrapper so series keeps progress totals without an explicit total. Generators have no cheap size — total stays omitted
  • Predicate throws are source errors: reported through onSourceError / sourceErrors, never onError
  • This is not intra-item abort: work already started on an item is not interrupted (use AbortController in your operation for that)

Usage Example:

import { series, stopWhen } from 'pipelean'

const {results} = await series(enrichOne, {
  total: albums.length,
  pause: ENRICH_DELAY_MS,
  onProgress: onItem,
})(stopWhen(albums, shouldStop))

// limit-style: close over counters instead of the item
stopWhen(folders, () => count >= limit)

forever

Purpose: Source adapter that yields the natural numbers 0, 1, 2, ... with no end — the unbounded counterpart of stopWhen. The building block for daemon loops: repeat a step until an abort predicate says so. Pipelean does not ship a daemon primitive; it ships the two halves, forever and stopWhen, and the composition is the whole story.

Type: () => asyncIterable<number> Sync: () => iterable<number>

Parameters: none.

Key Characteristics:

  • Unbounded: yields forever until the consumer stops pulling — every consumption must be bounded by stopWhen, take, or a break. A bare collect(forever()) never ends
  • Composable: stopWhen(forever(), predicate) is a predicate-stopped tick source; the predicate (usually an abort-signal check) is checked before each tick, so "cancel before work" holds
  • Index semantics: the yielded numbers are the tick indices a consumer receives — a cycle counter for free
  • Non-accumulating consumers recommended: a daemon has no items, so reduce/scan (stateful, no results array) fit better than series, which would otherwise rely on the undefined drop signal every tick

Usage Example:

import { reduce, rethrow, forever, stopWhen } from 'pipelean'

// daemon loop: one pass immediately, then repeat until Ctrl-C
const { value: progress } = await reduce(
  stopWhen(forever(), () => controller.signal.aborted),
  async (state, index) => {
    await runRound({ index })
    await sleep({ ms: everyMs, signal: controller.signal })
    return state
  },
  { cycles: 0, failures: 0 },
  { strategy: rethrow },
)

Composition

pipe

Purpose: Pipelean operation composer - chains functions left-to-right and preserves Pipelean's undefined drop signal.

Type: (...fns) => (input) => Promise<ReturnType<LastFn>>

Parameters:

  • Variadic arguments: Any number of sync or async functions to execute sequentially
  • input: The initial value passed to the first function

Return Type: A Promise that resolves to the final result.

Key Characteristics:

  • Functions execute left-to-right (first argument is applied to input)
  • Output of one function becomes input to the next
  • Supports both synchronous and asynchronous functions
  • Natural data flow from input through transformations
  • Designed for composing reusable operations passed to series() or used directly
  • Undefined Short-Circuit: If any step returns undefined, remaining steps are skipped and undefined is returned. This enables selection within a composed operation — see series drop behavior.

Usage Example:

import { pipe } from './functional.js'

const normalizeUser = pipe(
  async id => validateUserId(id),
  async id => fetchUser(id),
  user => user.active ? user : undefined,
  user => ({...user, email: user.email.toLowerCase()}),
)

const user = await normalizeUser(userId)

// Compose operations in a readable pipeline
const result = await pipe(
  async data => validate(data),
  async data => transform(data),
  async data => persist(data),
)(input)

Best Practice: Use pipe() when you need to chain operations that form a coherent data processing pipeline.

Selection in pipe (via undefined short-circuit):

import { pipe, series } from 'pipelean'

// Merge filter and transform in a single operation
const result = await series(items, pipe(
  x => x.active ? x : undefined,  // drop inactive items
  x => x.name,                     // extract name
))
// Items where active is false are skipped entirely

flow

Purpose: Sequentially apply multiple state-enrichment operations to one accumulated value. Each successful operation returns an object patch. Errors are handled per operation using Pipelean strategies.

Type: (operations, options?) => (initialState) => Promise<{value, errors, failure}>

Parameters:

  • operations: An array of functions. Each function operation(state) must return a non-null, non-array object (the patch to merge into the state). Operations can be sync or async.
  • options (optional):
    • strategy: Error strategy (default: collect). failFast, collect, failLate, skip, rethrow.
    • onError({operation, error, index, total}): Called for each handled operation error. Receives a normalized shape where operation is the function's name (or operation-${index} for anonymous functions) and total is the operation count.
    • onFailure(failure): Called when failure is truthy. Receives {operation, error, index} for failFast and {errors} for failLate — both normalized.

Return: A function (initialState) => Promise<{value, errors, failure}>.

The returned function:

  • Validates initialState (non-null, non-array object) and throws TypeError otherwise.
  • Executes each operation in order, passing the current accumulated state to each one.
  • Shallow-merges the returned patch into the state.
  • Treats an invalid operation return (null, array, primitive) as that operation throwing a TypeError, so the selected strategy applies.
  • Returns {value, errors, failure} where:
    • value: The final accumulated state. Even under failFast, this is the last successful accumulated state.
    • errors: Array of {operation, error, index} (normalized). Empty for skip and failFast.
    • failure: false for collect and skip; {operation, error, index} for failFast; {errors} for failLate. For rethrow, the original error is thrown instead of returning a result.

Key Characteristics:

  • Stateful accumulation: Each operation enriches the same state object. Output of one operation becomes the input to the next.
  • Shallow merge: {...state, ...patch} — top-level keys are overwritten, nested objects are not deep-merged.
  • Reusable pipeline: The pipeline is defined once. The returned function is called with different inputs.
  • Same error strategies as series / scan: All four strategies work, with the same onError / onFailure semantics. Default is collect.
  • No intermediate states: Only the final accumulated value is returned. There is no results array.
  • Empty operations array is valid and returns the initial state untouched.

Usage Example:

import { flow, collect, failFast } from 'pipelean'

const prepareAlbum = state => ({title: state.rawTitle.trim()})
const extractYear = state => ({year: parseYear(state.rawYear)})
const extractArtists = state => ({artists: state.artists ?? []})

const processAlbum = flow([
  prepareAlbum,
  extractYear,
  extractArtists,
])

const {value, errors, failure} = await processAlbum(input)
// value = {title, year, artists, rawTitle, rawYear, ...input}

// Strategy and callbacks are bound when the pipeline is defined
const processOrStop = flow(
  [prepareAlbum, extractYear, extractArtists],
  {
    strategy: failFast,
    onError: ({operation, error, index, total}) =>
      logger.error({operation, error, index, total}),
  },
)

const result = await processOrStop(input)

// Reuse the same pipeline with different inputs
const album1 = await processAlbum(rawAlbum1)
const album2 = await processAlbum(rawAlbum2)

When to use flow() vs pipe() vs series():

  • flow(): One input, many enrichments, final accumulated value. Stateful across the pipeline.
  • pipe(): Value-in / value-out composition. Each step transforms the previous step's return value. Stateless. Composable into series() and flow().
  • series(): Many inputs, one operation per input. Horizontal.

Error callback payload (always normalized):

  • onError({operation, error, index, total}) — operation is the function's name if set, otherwise operation-${index}. total is the operations array length.
  • onFailure({operation, error, index}) for failFast.
  • onFailure({errors}) for failLate, where each entry is {operation, error, index}.

join

Purpose: Fork/join over a record of named async tasks. Every branch starts immediately and join waits for all of them to settle, returning one structured result. Errors are data, per branch, using the same strategies as flow. This is not a concurrency runtime: no cancellation, no scheduler, no ordering guarantee.

Type: (tasks, options?) => Promise<{value, errors, failure}>

Parameters:

  • tasks: A non-null, non-array object mapping a branch name to a function task(key, index). Each function can be sync or async. The name is the stable key for that branch in the result.
  • options (optional):
    • strategy: Error strategy (default: collect). collect, failLate, skip, rethrow, and failFast.
    • onError({operation, error, index}): Called for each handled branch error. operation is the branch name.
    • onFailure(failure): Called (and awaited) when failure is truthy. Receives {errors} for failLate / failFast.

Return: Promise<{value, errors, failure}>:

  • value: Object mapping the successful branch names to their resolved values, in declaration order. A branch resolving undefined is still a success (value[key] === undefined).
  • errors: Array of {operation, error, index}, ordered by declaration (not completion). Empty for skip (and for rethrow, which throws instead).
  • failure: false for collect and skip; {errors} for failLate and failFast. For rethrow, the original error of the first failed branch is thrown instead of returning a result.

Throws (before any branch starts): TypeError if tasks is not a non-null, non-array object, if any value is not a function, or if it is a thenable (e.g. an un-awaited Promise).

Strategies:

  • collect (default): keep successes in value, failures in errors, failure: false.
  • failLate: like collect, then failure: {errors} if anything failed.
  • failFast (alias fail, stopOnError): behaves as failLate. Branches already started and cannot be stopped without an AbortSignal.
  • skip: errors stays empty, but onError still fires.
  • rethrow: throw the original error of the first failed branch, by declaration order, once every branch has settled.

Key Characteristics:

  • Named, heterogeneous branches: A record, not a collection. Each branch has a stable key, even when it fails: the failure is reported under errors with that operation.
  • All branches settle: join starts from a resolved promise, so a synchronous throw becomes a rejection; allSettled always captures it and nothing is left unhandled.
  • Completion order is irrelevant: value keys follow declaration order, and so does errors.
  • No drop signal: undefined is a legitimate branch value, unlike in series / pipe.
  • Empty record is valid: resolves to {value: {}, errors: [], failure: false}.
  • Not a concurrent map: for a homogeneous collection use p-map.

Usage Example:

import { join, failLate } from 'pipelean'

const ping = host => fetch(`https://${host}/health`).then(r => r.status)
const whoami = () => fetch('/whoami').then(r => r.json())

const { value, errors, failure } = await join({
  api: () => ping('api.example.com'),
  cdn: () => ping('cdn.example.com'),
  session: () => whoami(),
})

// value  = { api: 200, cdn: 200, session: {...} }
// errors = [] ; failure = false

// A failed branch stays named and leaves the rest intact
const down = await join({
  api: () => ping('api.example.com'),
  cdn: () => { throw new Error('unreachable') },
})
// down.value  = { api: 200 }
// down.errors = [{ operation: 'cdn', error: Error('unreachable'), index: 1 }]

// Treat any branch failure as a failure of the join
const strict = await join(
  { api: () => ping('api.example.com') },
  { strategy: failLate },
)

When to use join() vs series():

  • join(): A handful of named, heterogeneous tasks that run at the same time. One structured outcome, errors per branch.
  • series(): Many homogeneous items processed one at a time. Horizontal, ordered, with pause / take / onProgress.

Misc

assign

Purpose: Creates a flow() operation that conditionally assigns a property to state. If parse(state) returns undefined, the property is skipped (empty patch). Otherwise the property is set to the parsed value.

Type: (property, parse) => (state) => {} | {[property]: value}

Parameters:

  • property: The property name (string) to set on state.
  • parse: A function (state) => value | undefined. Return undefined to skip the property.

Return Type: A function suitable as a flow() operation. Returns {} when the parse is skipped, or {[property]: value} to merge into the accumulated state.

Key Characteristics:

  • Conditional assignment: Only sets the property when parse returns a non-undefined value.
  • Skip sentinel: Only undefined triggers skip — null, false, 0, "" are all assigned normally.
  • Composable: Multiple assign() calls compose naturally inside flow().

Usage Example:

import { assign, flow } from 'pipelean'

const extractName = assign('name', state => state.rawName.trim())
const extractYear = assign('year', state => {
  const n = Number.parseInt(state.rawYear, 10)
  return Number.isNaN(n) ? undefined : n
})

const {value} = await flow([extractName, extractYear])({
  rawName: '  Alice  ',
  rawYear: '1995',
})
// value = {rawName: '  Alice  ', rawYear: '1995', name: 'Alice', year: 1995}

where

Purpose: Creates a predicate function for strict equality object matching. Primarily used with filter for pattern-based selection.

Type: (pattern) => (item) => boolean

Parameters:

  • pattern: A plain object with key-value pairs to match against.

Return Type: A predicate function.

Usage Example:

import { where, filter } from 'pipelean'

const isAdmin = where({role: 'admin'})
const admins = await filter(users, isAdmin)

// Or use pattern directly with filter
const adults = await filter(users, {active: true})

retry

Purpose: Retry async functions with configurable attempts and delays between attempts.

Type: (fn, options) => retryFunction

Parameters:

  • fn: The async function to retry (required)
  • options: Configuration object with the following properties:
    • attempts: number (default: 3) - Number of retry attempts
    • delay: number (default: 0) - Delay between retry attempts in milliseconds

Behavior:

  • Retries on each attempt until successful or exhausted
  • Throws the last error after exhausting all attempts
  • No delay before first attempt
  • Applies configured delay between subsequent attempts

Usage Example:

import {retry} from 'pipelean'

const flakyOperation = async () => {
  if (Math.random() > 0.5)
    throw new Error('transient')
  return 'ok'
}

const robust = retry(flakyOperation, {attempts: 5, delay: 1000})
const result = await robust()

tryCatch

Purpose: Wraps individual async functions with lifecycle hooks for comprehensive error handling and telemetry. Always catches errors — returns null on failure.

Type: (fn, options) => wrapperFunction

Parameters:

  • fn: The async function to wrap
  • options: Configuration object with the following properties:
    • onStart: () => void — Called before function execution (no arguments)
    • onSuccess: (result) => void | Promise<void> — Called on successful completion (result only)
    • onError: (error) => void — Called on error (error only)
    • onFinally: () => void — Called regardless of success/failure (no arguments)

Return Type: Returns a wrapper function with the same signature as fn. Returns null on error.

Features:

  • Automatic async/await wrapping
  • Preserves original function signature
  • Comprehensive lifecycle: onStart → onSuccess/onError → onFinally
  • Deep telemetry support for debugging

Usage Example:

import { tryCatch } from 'pipelean'

const safeFetch = tryCatch(
  async (url) => {
    const response = await fetch(url)
    return await response.json()
  },
  {
    onError: (error) => {
      console.error('Fetch failed:', error)
    }
  }
)

const data = await safeFetch('https://api.example.com')
// data is the parsed JSON on success, null on error

Sync Functions

seriesSync, filterSync, findSync, scanSync, reduceSync, pipeSync, flowSync, and tryCatchSync are synchronous counterparts of the core functions.

They use the same error strategies and structured return conventions as the async iteration functions, but return values directly instead of wrapped in a Promise. Use them when your data and operations are synchronous and you still want Pipelean's structured error collection.

Available sync variants:

  • seriesSync returns {results, errors, sourceErrors, failure} directly
  • filterSync returns {results, errors, sourceErrors, failure} directly
  • findSync returns {result, errors, failure} directly and stops at the first match
  • scanSync returns {results, errors, sourceErrors, failure} directly
  • reduceSync returns {value, errors, sourceErrors, failure} directly
  • pipeSync composes synchronous functions left-to-right
  • flowSync returns {value, errors, failure} directly and runs a state-enrichment pipeline synchronously
  • tryCatchSync wraps a synchronous function with lifecycle hooks
  • stopWhenSync is the sync source adapter — same contract as stopWhen, sync iterables only

The sync variants handle source errors with the same semantics as their async twins: seriesSync, filterSync, scanSync, and reduceSync accept onSourceError({error, index}) and report errors thrown by the iteration itself in the sourceErrors field. See Source errors under series.

scanReduceSync is kept as an alias of reduceSync for backward compatibility.

Not supported in sync variants:

  • pause and pauseOnErrors, because they depend on async delay()
  • async iterables, because sync variants use plain synchronous iteration

Usage Example:

import { seriesSync, collect } from 'pipelean'

const {results, errors, failure} = seriesSync([1, 2, 3], x => {
  if (x === 2)
    throw new Error('bang')
  return x * 10
}, {strategy: collect})

// results: [10, 30]
// errors: [{item: 2, error: Error('bang'), index: 1}]
// failure: false

Early-exit selection with findSync:

import { findSync } from 'pipelean'

const users = [
  {id: 1, active: false},
  {id: 2, active: true},
  {id: 3, active: true},
]

const {result, errors, failure} = findSync(users, {active: true})

// result: {id: 2, active: true}
// errors: []
// failure: false

findSync supports the same predicate forms as filterSync, including object patterns via where(). It exits as soon as the first match is found. When no item matches, it returns {result: undefined, errors: [], failure: false}.

Synchronous state enrichment with flowSync:

import { flowSync, collect } from 'pipelean'

const processAlbumSync = flowSync([
  state => ({title: state.rawTitle.trim()}),
  state => ({year: parseYear(state.rawYear)}),
])

const {value, errors, failure} = processAlbumSync(input)
// value: the shallow-merged final state
// errors: []
// failure: false

flowSync mirrors flow() exactly but executes synchronously. Same strategies, same onError / onFailure semantics, same normalized error shapes. Use it when your operations and inputs are synchronous.

Related Documentation

  • README.md - Project overview and philosophy
  • guide.md - Comprehensive development guide with examples
  • examples.md - Practical usage examples for all functions