Comment on page
takeUntil
Consider a day at your workplace: You're anticipating an important email, but you've decided that once the clock hits 5 pm, you're clocking out, regardless of whether you've received that email or not. In this RxJS analogy, the anticipation of the email is one observable, while the 5 pm clock-out time is another. The
takeUntil operator ensures you're alert for the email's potential arrival, but the moment 5 pm arrives, you stop checking (unsubscribe).In real-world applications, think of a scenario where you're monitoring server responses on a dashboard. However, you want this monitoring to cease once a specific "Stop Monitoring" button is clicked. That's where
takeUntil shines.In the context of Angular,
takeUntil is particularly handy for auto-unsubscribing from observables when a component is destroyed. This is achieved by leveraging the ngOnDestroy lifecycle hook. You'd typically create a Subject, often named destroy$, and use it with takeUntil:private destroy$ = new Subject<void>();
observable$
.pipe(takeUntil(this.destroy$))
.subscribe(data => console.log(data));
ngOnDestroy() {
this.destroy$.next();
this.destroy$.complete();
}
With this setup, as soon as the
ngOnDestroy method is called (when the component is about to be destroyed), the observables using takeUntil with the destroy$ subject will automatically unsubscribe, ensuring that no unwanted memory leaks or unexpected behavior occurs.Example 1: Take values until timer emits
// RxJS v6+
import { interval, timer } from 'rxjs';
import { takeUntil } from 'rxjs/operators';
//emit value every 1s
const source = interval(1000);
//after 5 seconds, emit value
const timer$ = timer(5000);
//when timer emits after 5s, complete source
const example = source.pipe(takeUntil(timer$));
//output: 0,1,2,3
const subscribe = example.subscribe(val => console.log(val));
Example 2: Take the first 5 even numbers
// RxJS v6+
import { interval } from 'rxjs/observable/interval';
import { takeUntil, filter, scan, map, withLatestFrom } from 'rxjs/operators';
//emit value every 1s
const source = interval(1000);
//is number even?
const isEven = val => val % 2 === 0;
//only allow values that are even
const evenSource = source.pipe(filter(isEven));
//keep a running total of the number of even numbers out
const evenNumberCount = evenSource.pipe(scan((acc, _) => acc + 1, 0));
//do not emit until 5 even numbers have been emitted
const fiveEvenNumbers = evenNumberCount.pipe(filter(val => val > 5));
const example = evenSource.pipe(
//also give me the current even number count for display
withLatestFrom(evenNumberCount),
map(([val, count]) => `Even number (${count}) : ${val}`),
//when five even numbers have been emitted, complete source observable
takeUntil(fiveEvenNumbers)
);
/*
Even number (1) : 0,
Even number (2) : 2
Even number (3) : 4
Even number (4) : 6
Even number (5) : 8
*/
const subscribe = example.subscribe(val => console.log(val));
Example 3: Take mouse events on mouse down until mouse up
// RxJS v6+
import { fromEvent } from 'rxjs';
import { takeUntil, mergeMap, map } from 'rxjs/operators';
const mousedown$ = fromEvent(document, 'mousedown');
const mouseup$ = fromEvent(document, 'mouseup');
const mousemove$ = fromEvent(document, 'mousemove');
// after mousedown, take position until mouse up
mousedown$
.pipe(
mergeMap(_ => {
return mousemove$.pipe(
map((e: any) => ({
x: e.clientX,
y: e.clientY
})),
// complete inner observable on mouseup event
takeUntil(mouseup$)
);
})
)
.subscribe(console.log);
Last modified 1mo ago