Back in June 2024, I wrote a piece titled Advanced RxJs Operators You Know But Not Well Enough, which drew a lot of interest and proved helpful to many readers. Given that RxJS remains a cornerstone of Angular development and offers a huge toolbox of operators, I thought it was worth following up with a second edition. This time around, we'll look at a handful of operators, how they can be combined, and a few real-world situations where they shine.
The operators under the spotlight in this piece are:
forkJoin()vscombineLatest()auditTime()vsdebounceTime()pairwise()raceWith()iif()defer()
RxJS: forkJoin() vs combineLatest()
At first glance, these two operators seem to do the same thing: they both give you the last value emitted by a set of Observables.
The combineLatest operator works by emitting an array that holds the most recent values from each source Observable, but only after every one of them has produced at least a single value. From that point on, whenever any Observable in the group pushes out a new value, combineLatest() fires again with an updated array. One thing to watch out for is when you're dealing with a cold observable that has not yet emitted anything — in that case, your combineLatest() might stay silent indefinitely.
combineLatest([
this.stockPrice$, // emits stock prices
this.exchangeRate$ // emits exchange rates
]).subscribe(([stockPrice, exchangeRate]) => {
// will be logged every time any of the above observables emits
console.log(`Price: ${stockPrice}, Rate: ${exchangeRate}`);
});
The forkJoin operator, on the other hand, plays a waiting game. It listens to all provided Observables until they all complete, and only then does it release a single array containing the final value from each source. There is a catch: if any of the Observables throws an error or completes without emitting anything (i.e., returns EMPTY), then forkJoin() will mirror that behavior and either error out or complete with EMPTY as well. A common comparison is to Promise.all(), since both are designed to fire once, after all underlying operations have wrapped up.
forkJoin({
userProfile: this.api.getUserProfile(),
userSettings: this.api.getUserSettings(),
userPreferences: this.api.getUserPreferences()
}).subscribe(({ userProfile, userSettings, userPreferences }) => {
// will be logged only once, when all of the observables emits
console.log(userProfile, userSettings, userPreferences);
});
A pitfall I've seen more than once is someone plugging a WebSocket connection into a forkJoin. That's a recipe for trouble. The issue is that forkJoin sits and waits for every Observable to complete before it can emit, but WebSocket-driven observables are almost always hot — they keep pumping out values and rarely, if ever, complete on their own unless you explicitly unsubscribe from them.
forkJoin({
// API call (completes after fetching data)
apiData: this.http.get('/api/data'),
// WebSocket connection (never completes)
websocketData: this.websocketService.getUpdates()
}).subscribe(result => {
// will NEVER be logged
console.log('Result:', result);
});
RxJS: auditTime() vs debounceTime()
For a long time, I had a hard time telling these two apart. They look alike on the surface, but the small differences matter a great deal in practice.
Take debounceTime for example. This operator holds off on passing along a value from the source until the source has gone quiet for a set amount of time. It's your go-to when you want to wait for things to calm down — like a user pausing their typing — before you act on it.
Now, auditTime takes a different approach. It checks in on the source at fixed time intervals and, at the end of each interval, sends out the latest value it saw. This one is handy when you want regular snapshots of an ongoing stream of activity.
Most of us are comfortable slapping debounceTime on an input field to wait for the user to stop typing. But auditTime is arguably a better fit for things like tracking window resizes or scroll events, where there's a constant stream of updates. The example below shows exactly how these two behave differently when you resize the browser window. You'll see that auditTime keeps throwing out values the whole time the resize is happening, while debounceTime stays quiet until you've actually stopped.
// this will emit periodically
fromEvent(window, 'resize')
.pipe(
auditTime(500),
map(() => [window.innerWidth, window.innerHeight])
).subscribe((dimensions) => {
console.log(`AUDIT TIME:`, dimensions);
});
// this emits only when use stops the resizing
fromEvent(window, 'resize')
.pipe(
debounceTime(500),
map(() => [window.innerWidth, window.innerHeight])
).subscribe((dimensions) => {
console.log(`DEBOUNCE TIME:`, dimensions);
});
NOTE: A neat trick you can pull off with closures (or an injection token) is to create a function that hands you a signal, letting you listen in on window resize events from anywhere:
export const WINDOW_RESIZE_LISTENER =
new InjectionToken('Window resize listener', {
factory: () => {
const windowRef = inject(WINDOW);
return toSignal(
fromEvent(windowRef, 'resize').pipe(
auditTime(300),
map(() => windowRef.innerWidth),
startWith(windowRef.innerWidth),
takeUntilDestroyed(),
), { initialValue: windowRef.innerWidth });
},
});
Then, inside a component, you can use that injection token like this:
windowResize = inject(WINDOW_RESIZE_LISTENER);
// ^^ this is a signal
Beyond resizing, auditTime() comes in handy for game input handling or live data feeds — basically, any scenario where events are firing non-stop and you still want to run some logic at regular intervals.
RxJS: pairwise()
The pairwise operator is a transformation tool that takes the current value from an observable and pairs it up with the one that came before it, emitting the duo as [previous, current]. This is just the ticket when you need to see how values change from one emission to the next.
One practical application is keeping an eye on route changes to fine-tune navigation behavior.
import { Router, NavigationEnd } from '@angular/router';
import { filter, pairwise } from 'rxjs/operators';
this.router.events.pipe(
// filter for NavigationEnd events
filter(event => event instanceof NavigationEnd),
// pair consecutive route navigation events
pairwise()
).subscribe(([previous, current]: [NavigationEnd, NavigationEnd]) => {
console.log('Previous URL:', previous.url);
console.log('Current URL:', current.url);
});
Another classic use case is detecting which fields in a reactive form have actually been modified.
@Component({
imports: [ReactiveFormsModule],
})
export class FormTrackerComponent {
myForm = inject(FormBuilder).nonNullable.group({
name: '',
email: '',
});
constructor() {
// track changes in the form
this.myForm.valueChanges
.pipe(
// start with the initial form state
startWith(this.myForm.value),
// get the previous and current form values
pairwise(),
// get changed fields
map(([prev, curr]) => this.getChangedFields(prev, curr)),
// filter only distinct field keys
scan((acc, curr) =>
[...new Set([...acc, ...curr])], [] as string[]
)
)
.subscribe((fieldChange) => {
console.log('Changed fields:', fieldChange);
});
}
/**
* identify which fields have changed between two states.
* @returns - name of the field (name, email, age)
*/
private getChangedFields(previous: any, current: any): string[] {
return Object.keys(current).filter(
(key) => previous[key] !== current[key]
);
}
}
RxJS: race()
The race() operator is a bit of a gambler. It subscribes to several observables all at once, but as soon as the first one emits a value, it sticks with that winner (and keeps listening to it), immediately cancelling the subscriptions to all the others.
Honestly, race() isn't something I reach for every day, but recently I ran into a situation where it seemed like just the right tool. Picture this: you're making an API call to an endpoint that's acting up. The request could hang in the pending state forever, never resolving to either success or failure. What you'd ideally want is to give it a fixed amount of time, and if it's still stuck, bail out and show an error to the user. There are plenty of ways to approach this, but using race() was the first idea that came to mind:
@Component({ /* ... */ })
export class App {
#userAPIService = inject(UserAPIService);
displayItems = toSignal(race(
this.#userAPIService.getUsers().pipe(
map((data) => ({ status: 'loaded' as const, data}))
),
of({ status: 'failed' as const,}).pipe(delay(2000))
// ^^ emit failed status if no response after 2s
).pipe(startWith({ status: 'loading' as const})),
{ initialValue: { status: 'loading' } }
);
eff = effect(() => console.log(this.displayItemsSignal()));
}
So, the displayItem signal kicks off with a value of {status: 'loading'}, which lets you render a loading spinner on the page. Then it's a contest: either the getUsers() API call resolves in time, or if it's still hanging around after 2s, the {status: 'error'} value gets emitted and the pending API call is thrown away.
Granted, race() might feel a bit heavy-handed for this scenario. An alternative is to use the timeout() operator to achieve the same result, which would leave you with something like this:
displayItemsSignal = toSignal(
this.userAPIService.getUsers().pipe(
map((data) => ({
status: 'loaded' as const,
data,
})),
startWith({
status: 'loading' as const,
}),
timeout({
each: 1000,
with: () => of({ status: 'failed' as const }),
})
), { initialValue: { status: 'loading' } });
RxJS: defer()
The defer() operator is all about laziness — and that's a good thing. It lets you spin up a new observable only when someone actually subscribes to it. The logic inside doesn't even run until that subscription happens. It's not something you see used all the time, but if your codebase has a lot of Promises lying around, it could be a valuable tool to have in your belt.
To illustrate, imagine you have a service that makes API calls but, instead of the usual httpClient returning an Observable, it hands back a Promise.
@Injectable({ providedIn: 'root' })
export class UserAPIService {
#data = [{ name: 'user1' }, { name: 'user2' }, /*...*/];
getUsersPromise(): Promise<DataItem[]> {
return new Promise((res) =>
setTimeout(() => {
res(this.#data);
}, 200)
);
}
}
Now, the plan is to show a checkbox, and only when that checkbox is ticked should we actually go and fetch the users. You do a quick search on how to convert Promises into Observables and learn about the from operator, so you write something like the following:
@Component({
imports: [ReactiveFormsModule, AsyncPipe],
template: `
<label for="checkBox">check me</label>
<input type="checkbox" name="checkBox" [formControl]="control" />
@if(control.value){
@for(item of displayItems$ | async; track item.name){
{{ item.name }}
}
}
`,
})
export class App {
control = new FormControl<boolean>(false, { nonNullable: true });
displayItems$ = from(this.inject(UserAPIService).getUsersPromise());
}
Now, here's a detail I couldn't find highlighted in the rxjs from() docs: when you use from() with a Promise, it converts it immediately. This means your getUsersPromise() function is called right away, eagerly, without waiting for that checkbox to be clicked.
This might not always be a problem — after all, you probably wanted to load that user data anyway. It really just depends on whether eager loading is what you're after for that particular feature. If you'd rather hold off until the subscription is actually made (i.e., when the checkbox is clicked), you can wrap it with defer like so:
displayItems$ = defer(() => from(this.userAPIService.getUsersPromise()))
By using defer(), you effectively postpone the getUsersPromise() execution (and thus the API call) until the very moment someone subscribes.
RxJS: iif()
It's pretty common to find yourself in a situation where you're reacting to values from an observable, and inside a switchMap (or some other higher order observable) you use a conditional to decide what to return next. Here's a typical snippet:
displayItemsSignal = toSignal(
this.checkboxControl.valueChanges.pipe(
switchMap((isChecked) =>
isChecked
? this.userAPIService.getUsers()
: this.groupAPIService.getGroups()
)
), { initialValue: [] });
This works just fine, but there's a bit of syntactic sugar for this exact pattern. If both getUsers() and getGroups() return Observables, you can make use of the iif() operator to clean things up:
displayItemsSignal = toSignal(
this.checkboxControl.valueChanges.pipe(
switchMap((isChecked) =>
iif(
() => isChecked,
this.userAPIService.getUsers(),
this.groupAPIService.getGroups()
// ^^ both return an Observable of items
)
)
), { initialValue: [] });
There's a catch though. Let's tweak the scenario and say the service methods return Promises instead of Observables.
@Injectable({ providedIn: 'root' })
export class UserAPIService {
#data = [{ name: 'user1' }, { name: 'user2' }, /*...*/];
getUsersPromise(): Promise<DataItem[]> {
return new Promise((res) =>
setTimeout(() => {
console.log('UserAPIService resolved');
res(this.data);
}, 200)
);
}
}
@Injectable({ providedIn: 'root' })
export class GroupAPIService {
#data = [{ name: 'group1' }, { name: 'group2' }, /*...*/];
getGroupPromise(): Promise<DataItem[]> {
return new Promise((res) =>
setTimeout(() => {
console.log('GroupAPIService resolved');
res(this.data);
}, 200)
);
}
}
If you try to use iif() directly with those Promises, like this:
displayItemsSignal = toSignal(
this.checkboxControl.valueChanges.pipe(
switchMap((isChecked) =>
iif(
() => isChecked,
this.userAPIService.getUsersPromise(),
this.groupAPIService.getGroupPromise()
// ^^ both return a Promise of items
)
)
), { initialValue: [] });
...you're in for a surprise. No matter whether the checkbox is checked or not, both getUsersPromise() and getGroupPromise() will fire off. That's almost certainly not what you had in mind.
There are two straightforward ways to fix this. The first is to go back to the trusty ternary operator:
displayItemsSignal = toSignal(
this.checkboxControl.valueChanges.pipe(
switchMap((isChecked) =>
isChecked
? this.userAPIService.getUsersPromise()
: this.groupAPIService.getGroupPromise()
)
), { initialValue: [] });
This solves the issue when dealing with Promises. But if you're keen on sticking with the iif() operator, then you should pair it with defer() to prevent that eager execution, like so:
displayItemsSignal = toSignal(
this.checkboxControl.valueChanges.pipe(
switchMap((isChecked) =>
iif(
() => isChecked,
defer(() => this.userAPIService.getUsersPromise()),
defer(() => this.groupAPIService.getGroupPromise())
// ^^ delay the promise execution only if subscription happens
)
)
), { initialValue: [] });
So, to wrap it up: feel free to use the iif() operator whenever you're working with Observables. But when Promises are in the mix, you've got two options — revert to the Conditional (ternary) operator, or combine iif() with the defer() operator to keep things lazy.
Summary
As a follow-up to the earlier article, I set out to highlight operators that either tend to trip people up or prove their worth in niche situations. I hope you found this useful, and I'd love to hear your thoughts. Feel free to reach out on dev.to or connect with me on LinkedIn.
