RxJS-Lite Pipeable Operators

A reactive stream is a producer of many values over time — clicks, keystrokes, socket messages — that you consume by subscribing, and pipeable operators are the small, composable functions you chain to transform that stream before it reaches you. This is the model behind RxJS: you take a source and pass it through source.pipe(filter(...), map(...), debounceTime(...)), where each operator is a plain function that takes one Observable and returns a new one. You will build a lite version — the Observable plus enough operators to see how composition, cancellation, and timing hold together.

Export one object, rxjsLiteOperators. This is deliberately different from a fluent, method-chaining Observable: here operators are standalone functions combined with pipe, and the set includes time-based and higher-order operators that make the wiring genuinely hard.

Signature

// One exported object:
rxjsLiteOperators = {
  Observable,   // cold; new Observable(subscribeFn), subscribeFn may return a teardown
  of, from,     // creation: emit values / an iterable, then complete
  map, filter,  // transform / gate each value
  take,         // first n values, then complete AND unsubscribe from the source
  scan,         // running accumulation with a seed; emit each intermediate result
  debounceTime, // emit only after `ms` of silence; a new value resets the timer
  switchMap,    // map each value to an inner Observable and flatten; cancel the previous inner
};

// Observable instance:
//   .subscribe(observer | nextFn) -> { unsubscribe() }   // observer: { next, error, complete }
//   .pipe(...operators)           -> Observable          // pipe(a, b) === b(a(obs)); pipe() === obs
//
// Each operator is a standalone function:
//   map(fn) -> (sourceObservable) => newObservable

Examples

const { of, filter, map } = rxjsLiteOperators;

const out = [];
of(1, 2, 3, 4)
  .pipe(
    filter((x) => x % 2 === 0),
    map((x) => x * 10),
  )
  .subscribe((v) => out.push(v));
// out === [20, 40]

const { of, scan, take } = rxjsLiteOperators;

const totals = [];
of(1, 2, 3, 4)
  .pipe(
    scan((acc, x) => acc + x, 0),
    take(3),
  )
  .subscribe((v) => totals.push(v));
// totals === [1, 3, 6]   (take stops after 3 and unsubscribes the source)

For the time-based case, debounceTime(30) over a source that emits a, b, c within a few milliseconds emits only c, and only after 30ms of silence.

Notes

• Cold and lazy — the subscribe function runs per subscribe; nothing happens until someone subscribes, and each subscriber gets an independent run.
• Operators return new Observables — an operator is (source) => newObservable that subscribes to its source when subscribed and forwards transformed values. It must not drain or mutate the source when it is created.
• Unsubscribe tears down — subscribe returns an object with unsubscribe; calling it must stop further values and run the producer's teardown (clear timers, remove listeners). take, debounceTime, and switchMap all depend on this.
• Real timers — debounceTime uses setTimeout, and the tests use real timers with short delays. Clear your timers on unsubscribe and on complete, or a value leaks out after the consumer is gone.
• switchMap cancels — when a new outer value arrives, unsubscribe the previous inner subscription before subscribing to the new one. Keeping every inner alive is mergeMap, a different operator.
• Out of scope — no schedulers, no retry or catch operators, and scan always receives an explicit seed. Spend your effort on the composition model.