In RxJS 7, how do you write a reusable custom operator, both by composing existing operators with pipe() and from scratch?

## What an operator is In RxJS 7 a **pipeable operator** is an ordinary function that takes an `Observable` and returns another one. Its type is `OperatorFunction<T, R>`, or `MonoTypeOperatorFunction<T>` when input and output types match, both exported from `'rxjs'`. What people call "the `map` operator" is really a **factory**: `map(fn)` returns the operator function that `.pipe()` then calls with the source. `source$.pipe(a, b, c)` is simply `c(b(a(source$)))`. Anything with that shape can go into `.pipe()`, including your own functions. ## Option 1: compose with pipe() The RxJS guide recommends extracting a recurring sequence into a new operator with the **standalone `pipe()` function** (not the `.pipe()` method): 1. write a factory that returns `pipe(op1, op2, ...)`; 2. give it a descriptive name and explicit types; 3. use it like any built-in operator. Because it is built from existing operators, it inherits their correct handling of errors, completion and unsubscription for free. Calling `pipe()` with no arguments returns an identity function, and `.pipe()` with no arguments returns the same Observable. ## Option 2: build from scratch When no combination of existing operators expresses the behaviour - a rare case - return a function that wraps the source in `new Observable`: - **subscribe to the source inside the subscriber function**, so each subscription to the result creates its own subscription to the source; - **forward all three notifications**: `next` after your logic, and `error` and `complete` unchanged unless the operator deliberately alters them; - **return a teardown** that unsubscribes from the source and clears any timers or listeners you created; - **keep mutable state inside the subscriber function**, not in the factory, so two subscribers do not share a counter or buffer. The official guide's example re-implements `delay` this way: it tracks timers per subscription, delays `complete` until pending timers have fired, forwards `error` immediately, and clears every timer in the teardown. ## Common mistakes in hand-written operators | Mistake | Symptom | |---|---| | Not forwarding `error` | the consumer never learns of a failure; the stream appears to hang | | Not forwarding `complete` | `reduce`, `toArray` and `lastValueFrom` downstream never finish | | No teardown | the source keeps running after the consumer unsubscribes | | State declared in the factory | two subscribers corrupt each other's counters | | Using `Observable.lift` | relies on an internal API that is not part of the public surface | The RxJS 7 breaking-changes notes say `lift` was never documented for end users and recommend writing operators that return `new Observable` instead. ## Typing - Annotate the factory's return type as `OperatorFunction<In, Out>`; the compiler then checks that the composed steps line up. - Use a generic `<T>` for operators that do not care about the value type, such as a logging operator, typed `MonoTypeOperatorFunction<T>`. - Keep the factory's parameters simple values or functions, so the operator reads like the built-ins. ## Choosing between the two - If the behaviour can be described as "do A, then B" with existing operators, compose - it is shorter, and error, completion and unsubscription handling come from battle-tested code. - If the operator needs its own timing, its own buffering or custom completion rules that no combination expresses, write it from scratch - and test the error, completion and unsubscribe paths explicitly. - If you are unsure, compose first; a from-scratch rewrite can come later without changing callers, since both forms have the same function type. ## Imports in RxJS 7 Since RxJS 7.2 every operator is exported from `'rxjs'` itself, and the RxJS guide describes the `'rxjs/operators'` entry point as deprecated. New code imports `map`, `filter`, `pipe` and `Observable` from `'rxjs'`. ## Testing a custom operator A custom operator is a pure function of its source, so it tests well: feed it a known source such as `of(...)`, collect the output, and assert on values and completion. Timing-dependent operators are best tested with virtual time, which is its own topic. ## Interview summary Compose with `pipe()` first; write from scratch only when necessary, and then forward all notifications, keep state per subscription, and clean up in the teardown.