defs is a module, which defines all special constants and convenience methods used in chain().

Its properties are mixed in or otherwise duplicated and re-exported in other modules like chain().

The example of usage:

import {none} from 'stream-chain/defs.js';
// const {none} = require('stream-chain/defs.js');

A generator already expresses two of the markers natively, so don't yield them — there is no point:

• defs.none — to skip an item, simply don't yield (continue; or a guard clause).
• defs.many(...) — to produce several values, yield each one (or yield* anIterable).

The same goes for the contents of a defs.many(...) array: plain values only — no nested many(...), and none of none, stop, or finalValue(...).

The other two are supported from a generator, because a generator cannot express them on its own:

• defs.stop — yield it to terminate the whole pipeline. A generator's own return; ends only that generator; stop ends everything (see stop).
• defs.finalValue(x) — yield it to emit x as the segment's final value, skipping the segment's remaining functions — something a plain generator cannot do (see final values).

Rule: once you issue stop or finalValue(...) from a generator, exit it. return; immediately and yield nothing else in that invocation — the signal means "this generator is done," so producing more values after it is a logic error.

none is a special value, which terminates the chain and produces no value.

// import chain, {none} from 'stream-chain';
// import {chain, none} from 'stream-chain';
// const {chain, none} = require('stream-chain');

import chain from 'stream-chain';
import {none} from 'stream-chain/defs.js';

// a filter
dataSource.pipe(chain([x => x * x, x => (x % 2 == 0 ? none : x), x => 2 * x + 1]));
// skips even values

// if dataSource produces: 1, 2, 3
// then the result will be: 3, 19

This is the definition of none:

const none = Symbol.for('object-stream.none');

Note on null/undefined: asStream() and chain() treat null and undefined as none because Node.js streams reserve these values for end-of-stream signaling. gen() and fun(), however, are general-purpose compositors and pass null/undefined through the pipeline like any other value. Use none explicitly if you want consistent skip behavior across all contexts.

stop is a special value, which terminates the chain, produces no value, and stops further processing. Usually, it is used to terminate potentially infinite generators.

// import chain, {stop} from 'stream-chain';
// const {chain, stop} = require('stream-chain');

import chain from 'stream-chain';
import {stop} from 'stream-chain/defs.js';

chain([
  function* () {
    for (let i = 0; ; ++i) yield i;
  },
  n => (n > 1000 ? stop : n)
]);
// a stream produces numbers from 0 to 1000 inclusively

This is the definition of stop:

const stop = Symbol.for('object-stream.stop');

How stop terminates. stop is a sentinel; internally the executor short-circuits by throwing a Stop exception, and how that surfaces depends on how you consume the pipeline:

• Stream pipelines (chain(), asStream(), asWebStream()) absorb the Stop and end the stream cleanly; release external resources through the stream lifecycle (destroy / 'close').
• Functional pipelines (gen(), fun(), and stream-chain/core consumed directly) let the Stop propagate to the consumer — wrap your for await in try/catch to release external resources and to tell an early stop from a natural end.

stop is interpreted only by stream-chain's own function executor (chain functions, gen() / fun() blocks, generators, and functions wrapped by the adapters). A foreign Transform does not run through it, so it does not interpret stop.

Helper functions to mark values as final. A final value stops the chain and returns its value.

Scope. finalValue is interpreted by stream-chain's function executor — in chain functions, gen() / fun() blocks, generators, and functions wrapped by asStream() / asWebStream(). It emits its payload and skips the remaining functions of the current segment. A foreign Transform does not run through the executor, so it treats a finalValue payload as ordinary data.

Used internally to mark a value as a final value. The definition:

const finalSymbol = Symbol.for('object-stream.final');

This is a helper factory function, which can be used in by chained functions. It returns a special value, which terminates the chain and uses the passed value as the result of the chain.

// import chain, {finalValue} from 'stream-chain';
// const {chain, finalValue} = require('stream-chain');

import chain from 'stream-chain';
import {finalValue} from 'stream-chain/defs.js';

dataSource.pipe(
  chain([
    [
      x => x * x,
      x => finalValue(x),
      x => 2 * x + 1 // will be skipped
    ]
  ])
);

// if dataSource produces: 1, 2, 3
// then the result will be: 1, 4, 9

isFinalValue(value) is a companion to finalValue(). It checks if a value was marked as final returning a standard truthy/falsy result.

// import chain, {finalValue, isFinalValue} from 'stream-chain';
// const {chain, finalValue, isFinalValue} = require('stream-chain');

import chain from 'stream-chain';
import {finalValue, isFinalValue} from 'stream-chain/defs.js';

dataSource.pipe(
  chain([
    x => {
      let result = finalValue(x);
      // ...
      if (isFinalValue(result)) {
        // do something
      } else {
        // do something else
      }
      // ...
    }
    // the rest of pipeline
  ])
);

getFinalValue(value) is a companion to finalValue() and isFinalValue(). Its argument should be a wrapped final value. Its return will be an unwrapped value.

// import chain, {finalValue, isFinalValue, getFinalValue} from 'stream-chain';
// const {chain, finalValue, isFinalValue, getFinalValue} = require('stream-chain');

import chain from 'stream-chain';
import {finalValue, isFinalValue, getFinalValue} from 'stream-chain/defs.js';

dataSource.pipe(
  chain([
    x => {
      let result = finalValue(42);
      // ...
      if (isFinalValue(result)) {
        const value = getFinalValue(result);
        console.log(value);
        // do something
      } else {
        console.log(result);
        // do something else
      }
      // ...
    }
    // the rest of pipeline
  ])
);

The right way to return multiple values is to use a generator function. Sometimes it is not possible to do that for some reason, e.g., because of performance considerations or for simplicity.

That's why there are helper functions that allow you to return multiple values from regular functions.

The obvious downside is that a generator function sends each value down the chain as soon as it is produced, while a regular function will accumulate the values in an array before they are sent down the chain as multiple values.

The other reason for this facility is historical: the stream-chain library was originally designed when generators were not available. Now it can be used for backward compatibility.

Used internally to mark a value as multiple values. The definition:

const manySymbol = Symbol.for('object-stream.many');

This is a helper factory function, which is used to wrap arrays to be interpreted as multiple values returned from a function.

// import chain, {many} from 'stream-chain';
// const {chain, many} = require('stream-chain');

import chain from 'stream-chain';
import {many} from 'stream-chain/defs.js';

dataSource.pipe(chain([x => many([x, x + 1, x + 2])]));

// if dataSource produces: 1, 5
// then the result will be: 1, 2, 3, 5, 6, 7

isMany(value) is a companion to many(). It checks if a value was marked as multiple values returning a standard truthy/falsy result.

// import chain, {many, isMany} from 'stream-chain';
// const {chain, many, isMany} = require('stream-chain');

import chain from 'stream-chain';
import {many, isMany} from 'stream-chain/defs.js';

dataSource.pipe(
  chain([
    x => {
      let result = many([x, x + 1]);
      // ...
      if (isMany(result)) {
        // do something
      } else {
        // do something else
      }
      // ...
    }
    // the rest of pipeline
  ])
);

getManyValues(value) is a companion to many() and isMany(). Its argument should be a wrapped multiple value. Its return will be an unwrapped value (an array of values).

// import chain, {many, isMany, getManyValues} from 'stream-chain';
// const {chain, many, isMany, getManyValues} = require('stream-chain');

import chain from 'stream-chain';
import {many, isMany, getManyValues} from 'stream-chain/defs.js';

dataSource.pipe(
  chain([
    x => {
      let result = many([1, 42, 99]);
      // ...
      if (isMany(result)) {
        const values = getManyValues(result);
        console.log(values);
        // do something
      } else {
        console.log(result);
        // do something else
      }
      // ...
    }
    // the rest of pipeline
  ])
);

Since 3.4.0

toMany(value) is a companion to many(). It converts a single value to a multiple value.

// import {none, toMany} from 'stream-chain';
// const {none, toMany} = require('stream-chain');

import {none, toMany} from 'stream-chain/defs.js';

toMany(none); // => many([])
toMany(42); // => many([42])
toMany(many([1, 2, 3])); // => many([1, 2, 3])

toMany() returns a Many value.

Since 3.4.0

normalizeMany(value) is a companion to many(). It converts a multiple value to a single value, if possible:

// import {many, normalizeMany} from 'stream-chain';
// const {many, normalizeMany} = require('stream-chain');

import {many, normalizeMany} from 'stream-chain/defs.js';

normalizeMany(many([])); // => none
normalizeMany(many([1])); // => 1
normalizeMany(many([1, 2, 3])); // => many([1, 2, 3])

normalizeMany() returns none, if the value is an empty Many value. If it is a Many value with a single value, it returns that value. Otherwise, it returns the original value.

Since 3.4.0

combineMany(...args) is a companion to many(). It takes any number of arguments and returns a Many value containing all values from all arguments:

// import {many, combineMany} from 'stream-chain';
// const {many, combineMany} = require('stream-chain');

import {many, combineMany} from 'stream-chain/defs.js';

combineMany(none, none); // => many([])
combineMany(none, 2); // => many([2])
combineMany(1, none); // => many([1])
combineMany(none, many([])); // => many([])
combineMany(1, many([])); // => many([1])
combineMany(none, many([1, 2, 3])); // => many([1, 2, 3])
combineMany(0, many([1, 2, 3])); // => many([0, 1, 2, 3])
combineMany(many([1]), many([2])); // => many([1, 2])
combineMany(1, 2, 3); // => many([1, 2, 3])
combineMany(many([1]), 2, many([3, 4])); // => many([1, 2, 3, 4])

combineMany() returns a new Many value. Arguments are unchanged. Values are concatenated in the order they are provided.

Since 3.4.0

combineManyMut(a, ...args) is a companion to many(). It takes a required first argument and any number of additional arguments, returning a Many value containing all values. This function is like combineMany(), but it can mutate a if it is a Many by appending values from subsequent arguments. Only a may be modified; the rest are never mutated.

This function is provided for performance reasons and should be used with caution.

Example:

// import {many, combineManyMut} from 'stream-chain';
// const {many, combineManyMut} = require('stream-chain');

import {many, combineManyMut} from 'stream-chain/defs.js';

let result = many([1, 2, 3]);
if (someFlag) {
  result = combineManyMut(result, many([4, 5, 6]));
}
// do something with result

Helper functions to mark a function as being flushable. A flushable function will be called when the end of the stream is reached. When it happens, it will be called with a special value none (see above). This value is never used in the normal course of processing. When it happens, the function should produce delayed values, if any.

Unmarked functions are not flushable and will not be called when the end of the stream is reached.

Used internally to mark a function as being flushable. The definition:

const flushSymbol = Symbol.for('object-stream.flush');

This function marks a function as being flushable. Its arguments are:

• fn is the function to be marked as flushable.
• final is an optional function that will be called when the end of the stream is reached with no arguments. Otherwise, fn will be called with a single argument none (see above). Defaults to null.

It returns the augmented/modified function.

// import chain, {none, flushable} from 'stream-chain';
// const {chain, none, flushable} = require('stream-chain');

import chain from 'stream-chain';
import {none, flushable} from 'stream-chain/defs.js';

let acc = 0;

dataSource.pipe(
  chain([
    flushable(x => {
      if (x === none) {
        return acc; // return the accumulated value
      }
      acc += x;
      return none; // produce no result
    })
  ])
);
// if dataSource produces: 1, 2, 3
// then the result will be: 6

The same example can be reformulated like that:

let acc = 0;

dataSource
  .pipe(chain([
    flushable(
      x => void acc += x,
      () => acc
    )
  ]));
// if dataSource produces: 1, 2, 3
// then the result will be: 6

isFlushable(fn) is a companion to flushable(). It checks if a function is marked as flushable. It is mostly used internally.

Helper functions to mark a function as being a wrapper for a list of functions. A function list is an array of functions used in chains. It is a feature used internally to optimize pipelines: when available a function list is inlined into the pipeline instead of using the wrapper function. It is used by gen and fun modules.

This optimization is used for performance reasons and is enabled by default.

In some cases this it can cause logical problems. For example, if you use finalValue() it will not work as expected if you use a function list. Essentially it will be globalized. If your intention was to confine the final value to a specific segment, use clearFunctionList() to suppress the optimization.

Used internally to mark a function as being derived from a function list. The definition:

const fListSymbol = Symbol.for('object-stream.fList');

This function marks a value as being derived from a function list. Its arguments are:

• value is the value to be marked, usually a function.
• fns is an array of functions.

It returns the augmented/modified value.

This function checks if a value is derived from a function list. It is mostly used internally.

This function extracts the function list from a value. It is mostly used internally.

In some cases we want to suppress the function list optimization and use the wrapper function as is. A good example would be final values scoped for a specific function list. This function clears the function list from a value suppressing the optimization.

It returns fn.

Example:

import {gen, clearFunctionList} from 'stream-chain';

// treated as a wrapper, the functions will be inlined
const p1 = gen(
  x => x + 1,
  x => x * x
);

// treated as an opaque function
const p2 = clearFunctionList(
  gen(
    x => x + 1,
    x => x * x
  )
);

const c1 = gen(p1, p2);
// effectively the same as:
const c2 = gen(
  x => x + 1,
  x => x * x,
  p2
);

Shape-based type predicates for distinguishing Node Streams from Web Streams. None of them import node:stream — all are pure duck-typing on method/state presence, so the Web guards are browser-safe to call. Used internally by chain() to dispatch incoming stream objects to the right adapter; exported for downstream consumers who need the same dispatch.

import {isReadableWebStream, isWritableWebStream, isDuplexWebStream} from 'stream-chain/defs.js';

• isReadableWebStream(x) — true if x looks like a Web Streams ReadableStream (getReader + pipeTo).
• isWritableWebStream(x) — true if x looks like a Web Streams WritableStream (getWriter + abort).
• isDuplexWebStream(x) — true if x is a {readable, writable} pair (the shape returned by TransformStream and asWebStream).

import {isReadableNodeStream, isWritableNodeStream, isDuplexNodeStream} from 'stream-chain/defs.js';

• isReadableNodeStream(x) — true if x looks like a Node Readable (or Duplex/Transform). Checks for .pipe/.on/_readableState.
• isWritableNodeStream(x) — true if x looks like a Node Writable (or Duplex/Transform). Checks for .write/.on/_writableState.
• isDuplexNodeStream(x) — true if x is a Node Duplex/Transform.

Node guards live in defs.js (not node:stream-importing code) so they can be called from any substrate without pulling Node Streams into a browser bundle.

Used to terminate the processing of a value by a pipeline. It can be thrown by any function in the pipeline.

Usually returning the stop value is more convenient than throwing it.

// import chain, {Stop} from 'stream-chain';
// const {chain, Stop} = require('stream-chain');

import chain from 'stream-chain';
import {Stop} from 'stream-chain/defs.js';

chain([
  function* () {
    for (let i = 0; ; ++i) yield i;
  },
  n => {
    if (n > 1000) throw new Stop();
    return n;
  }
]);
// a stream produces numbers from 0 to 1000 inclusively