Callbag Pipe Utility

1.2.0 · active · verified Sun Apr 19

callbag-pipe is a lightweight utility function designed to chain callbags together in a functional pipeline, conceptually similar to `Ramda.pipe` or `lodash.flow`. It serves as a foundational component within the Callbag ecosystem, facilitating the clear and concise composition of Callbag sources and operators. The current stable version is 1.2.0, and as a small, focused utility, it typically sees a low-frequency release cadence, with updates primarily for compatibility or minor enhancements rather than new features. Its key differentiator is its seamless integration with the Callbag specification, making it a natural choice for creating and consuming Callbag streams, while offering a familiar functional composition pattern. It provides TypeScript types for improved developer experience and type safety.

Common errors

```
TypeError: pipe is not a function
```
cause The `pipe` function was imported incorrectly, or the module was not properly resolved in the environment.

fix Ensure you are using the correct import syntax for your module system: `import { pipe } from 'callbag-pipe';` for ESM or `const pipe = require('callbag-pipe');` for CommonJS. Check your `package.json` for proper module configuration if using bundlers.
```
Argument of type '...' is not assignable to parameter of type 'Callbag<any, any>'.
```
cause Type mismatch when chaining Callbag operators within `pipe`, often due to an operator expecting a different type of source or producing an unexpected output type.

fix Review the type signatures of the Callbag operators being chained. Ensure that the output type of one operator matches the expected input type of the next. Use explicit type annotations in TypeScript if necessary to debug type flow, e.g., `map(([x, y]: [number, number]) => ...)`.

Warnings

gotcha When nesting `pipe` calls to create new operators or inline pipelines, the inner `pipe` must explicitly receive its source argument as a function. Direct nesting like `pipe(source, pipe(op1, op2))` will not work as expected.
Fix: Wrap the inner `pipe` in a function that accepts the source. For example, `s => pipe(s, map(f), take(amount))` or `s => pipe(s, ...innerOperators)`.
gotcha Although `callbag-pipe` itself has no runtime dependencies outside of the Callbag specification, its utility is realized by composing it with other Callbag operators (e.g., `callbag-map`, `callbag-take`). Ensure these operators are installed and imported correctly when building pipelines.
Fix: Install necessary Callbag operators via `npm install callbag-map callbag-take` etc., and import them appropriately (`import map from 'callbag-map';`).

Install

npm install callbag-pipe npm
yarn add callbag-pipe yarn
pnpm add callbag-pipe pnpm

Imports

pipe
wrong:
```
import pipe from 'callbag-pipe';
```
correct:
```
import { pipe } from 'callbag-pipe';
```
The `pipe` function is typically exported as a named export in ESM environments.
pipe (CommonJS)
```
const pipe = require('callbag-pipe');
```
For CommonJS environments, `pipe` is the default export of the module.
pipe (Type Definition)
```
import type { pipe } from 'callbag-pipe';
```
When only importing types, use the `type` keyword for tree-shaking and clarity, especially when not directly using the runtime value in a TypeScript project.

Quickstart

This example demonstrates creating a complete Callbag stream from source to sink using `pipe`, combining two interval streams, mapping their values, taking a limited number, and logging the output.

import { pipe } from 'callbag-pipe';
import interval from 'callbag-interval';
import forEach from 'callbag-for-each';
import combine from 'callbag-combine';
import take from 'callbag-take';
import map from 'callbag-map';

// Create a pipeline that combines two intervals, maps their values, takes the first 10, and logs them.
pipe(
  combine(interval(100), interval(350)),
  map(([x, y]: [number, number]) => `X${x},Y${y}`),
  take(10),
  forEach((x: string) => console.log(x))
);

// Expected output:
// X2,Y0
// X3,Y0
// X4,Y0
// X5,Y0
// X6,Y0
// X6,Y1
// X7,Y1
// X8,Y1
// X9,Y1
// X9,Y2

view raw JSON →