RxJS — One Operator to flatten Them All
A new experimental flattening operator may be coming to rxjs called flatmap.
When I first heard this, I thought “Isn’t flatMap just a deprecated name for mergeMap?” and I was right:
So if it’s obviously *not* that, then what is it?
RxJs core team lead Ben Lesh opened a discussion on an idea for an operator that could serve as a one-stop shop for all your observable flattening needs.
Instead of dealing with individual operators like switchMap(fn) or exhaustMap(fn), the proposal introduces a unified syntax:
declare const flatMap = (
project: (value: In, index: number) => ObservableInput<Out>,
config?: { concurrency?: number; behavior?: "merge" | "switch" | "exhaust" },
) => OperatorFunction<In, Out>;
// behavior defaults to "merge"
// concurrency defaults to 1
flatMap(fn); // concatMap(fn)
flatMap(fn, { concurrency: Infinity }); // mergeMap(fn)
flatMap(fn, { behavior: 'switch' }); // switchMap(fn)
flatMap(fn, { behavior: 'exhaust' }); // exhaustMap(fn)While this may seem unremarkable at first glance, the true excitement lies in the possibilities that this operator could unlock, previously unattainable with existing flattening operators:
Concurrent Switching 🤯
Enabling concurrent switching involves interrupting the “oldest” subscription once the concurrency limit is reached:
flatMap(fn, { concurrency: 3, behavior: 'switch' })
In this example, as soon as we go over the concurrency limit, the initial subscription (foo.txt) is stopped.
Concurrent Exhausting 🤯
Concurrent exhausting entails ignoring incoming subscriptions when the concurrency limit is met:
flatMap(fn, { concurrency: 3, behavior: 'exhaust' })
As soon as we hit our concurrency limit of 3, all attempts to create a new subscription are ignored.
Summary
Consolidating observable flattening operations under a unified term, like “flatMap,” helps achieve clarity, but beyond simplifying nomenclature, the flatMap operator adds useful functionalities to existing flattening behaviors. I’m anticipating its integration into future RxJS versions and, who knows, maybe even as a method of native observables. 🤞
