There’s no denying that the extensive collection of operators is what makes the RxJS library such a formidable asset for any developer. Not long ago, I found myself writing multiple custom operators to neatly factor out recurring patterns of operator chains. In what follows, I’ll walk you through the basics of crafting your own RxJS operators and show you a few practical examples.
Identity operator
At its core, an RxJS operator boils down to a function that accepts a source observable and returns a resulting stream. So, when it comes to building a custom RxJS operator, all you need to do is write a standard JavaScript (TypeScript) function. Take, for instance, a basic identity operator that simply passes through whatever the source emits:
import { interval, Observable } from "rxjs";
import { take } from "rxjs/operators";
const source$ = interval(1000).pipe(take(3));
function identity<T>(source$: Observable<T>): Observable<T> {
return source$;
}
const results$ = source$.pipe(identity);
results$.subscribe(console.log);
// console output: 0, 1, 2
Now we’ll introduce minimal functionality into a bespoke operator.
Logging operator
Below, a bespoke operator triggers a side effect—printing every emitted value of the source stream to the console:
import { interval, Observable } from "rxjs";
import { take, tap } from "rxjs/operators";
const source$ = interval(1000).pipe(take(3));
function log<T>(source$: Observable<T>): Observable<T> {
return source$.pipe(tap(v => console.log(`log: ${v}`)));
}
const results$ = source$.pipe(log);
results$.subscribe(console.log);
// console output: log: 0, log: 1, log: 2
The source$ stream forms the foundation of the resulting observable, which is then modified through the application of native operators using the pipe method.
Factory for operators
When you need to give a custom operator some surrounding context, defining a function that yields that operator proves handy. Any parameters passed to the factory become part of the operator’s closure.
import { interval, Observable } from "rxjs";
import { take, tap } from "rxjs/operators";
const source$ = interval(1000).pipe(take(3));
function logWithTag<T>(tag: string): (source$: Observable<T>) => Observable<T> {
return source$ =>
source$.pipe(tap(v => console.log(`logWithTag(${tag}): ${v}`)));
}
const results$ = source$.pipe(logWithTag("RxJS"));
results$.subscribe(console.log);
// console output: logWithTag(RxJS): 0, logWithTag(RxJS): 1, logWithTag(RxJS): 2
By leveraging the MonoTypeOperatorFunction from RxJS, you can cut down on the verbosity of your return type signature. Furthermore, defining an operator becomes more concise with the static pipe function:
import { interval, MonoTypeOperatorFunction, pipe } from "rxjs";
import { take, tap } from "rxjs/operators";
const source$ = interval(1000).pipe(take(3));
function logWithTag<T>(tag: string): MonoTypeOperatorFunction<T> {
return pipe(tap(v => console.log(`logWithTag(${tag}): ${v}`)));
}
const results$ = source$.pipe(logWithTag("RxJS"));
results$.subscribe(console.log);
// console output: logWithTag(RxJS): 0, logWithTag(RxJS): 1, logWithTag(RxJS): 2
Looking for additional RxJS insights? Take a look at this.
Observer-unique lexical scope
The factory function behind an operator gets invoked exactly once, when the stream is first defined. This means every observer that subscribes to that stream ends up sharing a single lexical scope:
import { interval, MonoTypeOperatorFunction, pipe } from "rxjs";
import { take, tap } from "rxjs/operators";
const source$ = interval(1000).pipe(take(3));
function tapOnce<T>(job: Function): MonoTypeOperatorFunction<T> {
let isFirst = true;
return pipe(
tap(v => {
if (!isFirst) {
return;
}
job(v);
isFirst = false;
})
);
}
const results$ = source$.pipe(tapOnce(() => console.log("First value emitted")));
results$.subscribe(console.log);
results$.subscribe(console.log);
// console output: First value emitted, 0, 0, 1, 1, 2, 2
To obtain a distinct lexical context per observer, the defer utility comes in handy:
import { defer, interval, MonoTypeOperatorFunction } from "rxjs";
import { take, tap } from "rxjs/operators";
const source$ = interval(1000).pipe(take(3));
function tapOnceUnique<T>(job: Function): MonoTypeOperatorFunction<T> {
return source$ =>
defer(() => {
let isFirst = true;
return source$.pipe(
tap(v => {
if (!isFirst) {
return;
}
job(v);
isFirst = false;
})
);
});
}
const results$ = source$.pipe(tapOnceUnique(() => console.log("First value emitted")));
results$.subscribe(console.log);
results$.subscribe(console.log);
// console output: First value emitted, 0, First value emitted, 0, 1, 1, 2, 2
Curious about an alternative approach to tackling the tapOnce challenge? Check out this post on my blog.
Practical use cases
Here’s the firstTruthy operator:
import { MonoTypeOperatorFunction, of, pipe } from "rxjs";
import { first } from "rxjs/operators";
const source1$ = of(0, "", "foo", 69);
function firstTruthy<T>(): MonoTypeOperatorFunction<T> {
return pipe(first(v => Boolean(v)));
}
const result1$ = source1$.pipe(firstTruthy());
result1$.subscribe(console.log);
// console output: foo
evenMultiplied operator:
import { interval, MonoTypeOperatorFunction, pipe } from "rxjs";
import { filter, map, take } from "rxjs/operators";
const source2$ = interval(10).pipe(take(3));
function evenMultiplied(multiplier: number): MonoTypeOperatorFunction<number> {
return pipe(
filter(v => v % 2 === 0),
map(v => v * multiplier)
);
}
const result2$ = source2$.pipe(evenMultiplied(3));
result2$.subscribe(console.log);
// console output: 0, 6
liveSearch operator:
import { ObservableInput, of, OperatorFunction, pipe } from "rxjs";
import { debounceTime, delay, distinctUntilChanged, switchMap } from "rxjs/operators";
const source3$ = of("politics", "sport");
type DataProducer<T> = (q: string) => ObservableInput<T>;
function liveSearch<R>(
time: number,
dataProducer: DataProducer<R>
): OperatorFunction<string, R> {
return pipe(
debounceTime(time),
distinctUntilChanged(),
switchMap(dataProducer)
);
}
const newsProducer = (q: string) =>
of(`Data fetched for ${q}`).pipe(delay(2000));
const result3$ = source3$.pipe(liveSearch(500, newsProducer));
result3$.subscribe(console.log);
// console output: Data fetched for sport
Conclusions
When you find yourself repeatedly assembling the same RxJS operators, you can bundle them into a single custom operator and apply that everywhere the same logic is needed. By leveraging generics, subsequent operators in the chain can retain precise typing for the values that pass through.
See the runnable demo:
Thanks for reading — I trust it gave you a fresh takeaway.