Pipelines

#Why pipelines

Deriving one signal from another is just a computed. But once you want a few steps in sequence, project a shape, drop duplicate values, keep a running total, the computed body grows, and steps you would like to reuse get inlined and copied around.

If you come from RxJS the instinct is to reach for a pipe of operators. Pipelines give you that shape without leaving signals: no subscription to manage, no effect() bridging a stream back into a signal. Each operator is a small Signal to Signal function, and .pipe(...) threads them together into one derived signal that stays synchronous and glitch-free.

Reach for a pipeline when a derivation is more than a one-liner and you would rather read it as a sequence of named steps than as a dense computed.

#piped and pipeable

piped(initial) creates a writable signal that already carries the .pipe(...) and .map(...) API. pipeable(existing) adds the same API to a signal you already have. Both keep the underlying signal writable, so the source end stays a normal WritableSignal.

import { piped, pipeable } from '@mmstack/primitives';

const count = piped(1);

// .map composes value -> value transforms inline
const label = count.map(
  (n) => n * 2,
  (n) => `#${n}`,
);

// .pipe retrofits onto an existing signal too
const view = pipeable(signal(0)).pipe(/* operators */);

.map(...) is the quick path: pass one or more inline value-to-value functions and they run left to right. .pipe(...) is the same idea with the named operators below, where each one is a reusable Signal-to-Signal transform. The result of both is itself pipeable, so you can keep chaining.

#A real pipeline

Say you have a stream of scroll deltas and want a distinct running total, tripled, without recomputing when the value has not actually changed. Read top to bottom: multiply, drop repeats, accumulate.

import { pipeable, map, distinct, scan } from '@mmstack/primitives';

const total = pipeable(signal(10)).pipe(
  map((n) => n * 3),
  distinct(),
  scan((acc, n) => acc + n, 0),
);

Every operator has the shape (src: Signal<I>) => Signal<O>, so the output type of one is the input type of the next and the chain stays fully typed.

#Operators by purpose

There are three groups. Projection reshapes a value: map(fn) is a pure transform, and select(fn, opt?) is the same with a CreateSignalOptions bag so you can pass a custom equal or debugName through to the underlying computed. combineWith(other, fn) projects the piped signal together with a second signal and recomputes when either changes.

import { pipeable, select, combineWith } from '@mmstack/primitives';

const user = pipeable(signal({ id: 1, name: 'Ada' }));

// compare by id so unrelated field changes don't re-emit
const name = user.pipe(select((u) => u.name, { equal: (a, b) => a === b }));

const qty = signal(3);
const total = pipeable(signal(10)).pipe(combineWith(qty, (price, q) => price * q));

Gating controls which values pass. distinct(equal?) suppresses an emission when equal(prev, next) is true (defaults to Object.is), which is how you compare by id and ignore noisy fields. filter(predicate) keeps only passing values, holding the last one and returning undefined until the first match; filterWith(predicate, initial) is the same but seeds an initial value so the type never includes undefined.

import { pipeable, filter, distinct } from '@mmstack/primitives';

const event = pipeable(signal<MouseEvent | null>(null));

// undefined until the first click, then holds the last click
const lastClick = event.pipe(filter((e): e is MouseEvent => e?.type === 'click'));

// re-emit only when id changes, ignore noisy fields
const stable = user.pipe(distinct((a, b) => a.id === b.id));

Stateful operators remember across emissions. scan(reducer, seed) folds each value into a running accumulator, like Array.prototype.reduce over time. pairwise() emits [prev, curr] tuples (prev is undefined on the first). startWith(initial) emits a seed first, then mirrors the source, handy for a loading value before real data arrives.

One operator sits apart. tap(fn, injector?) runs a side effect through effect() and passes the value along unchanged, for logging or analytics. Because it uses effect() it needs an injection context, so pass an Injector when you build the pipeline outside one, and do not use it to write other signals.