Rxjs is a library for doing reactive programming. Creation operators are useful for generating data from various data sources to be subscribed to by Observers.
In this article, we’ll look at more filtering operators, including skipUntil, skipWhile, take, takeLast, takeUntil, and takeWhile operators.
skipUntil
The skipUntil operator returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.
It takes one argument, which is the notifer Observable. When this emits an item, then the source Observable’s emitted values will be emitted by the skipUntil operator.
For example, we can use it as follows:
import { interval } from "rxjs";
import { skipUntil, take } from "rxjs/operators";
const numbers = interval(1000).pipe(
take(10),
skipUntil(interval(5000).pipe(take(1)))
);
numbers.subscribe(x => console.log(x));
The code above takes the emitted values from the interval(1000) Observabkem then pipe it to the take(10) operator to take the first 10 values, then take those values to the skipUntil operator, which has the interval(5000).pipe(take(1)) Observable passed in.
interval(5000).pipe(take(1)) emits a value after 5 seconds, which then will trigger the emission of the interval(1000) Observable’s emitted values.
Then we should see:
4
5
6
7
8
9
skipWhile
The skipWhile operator returns an Observable that skips all items emitted by the source Observable until the condition returned by the predicate function becomes false.
It takes one argument, which is a predicate function to test each item with the condition returned by the function.
For example, we can write the following:
import { of } from "rxjs";
import { skipWhile } from "rxjs/operators";
const numbers = of(1, 2, 3, 4, 5, 6).pipe(skipWhile(x => x < 3));
numbers.subscribe(x => console.log(x));
Then the items from the of(1, 2, 3, 4, 5, 6) Observable won’t be emitted when the value is less than 3. This means that numbers will emit 3, 4, 5 and 6.
Once the condition in the predicate function is tested false once, it’ll keep emitting regardless whether the emitted value from the source Observable make the condition evaluate to false or not.
So if we have:
import { of } from "rxjs";
import { skipWhile } from "rxjs/operators";
const numbers = of(1, 2, 3, 4, 3, 2).pipe(skipWhile(x => x < 3));
numbers.subscribe(x => console.log(x));
Then we get 3, 4, 5 and 2 since the condition becomes false once the first 3 was emitted by of(1, 2, 3, 4, 3, 2).
take
The take operator returns an Observable that emits the first count values emitted by the source Observable.
It takes one argument which is the count, which the maximum number of values to emit.
For example, we can use it as follows:
import { of } from "rxjs";
import { take } from "rxjs/operators";
const numbers = of(1, 2, 3, 4, 3, 2).pipe(take(2));
numbers.subscribe(x => console.log(x));
Then we get 1 and 2 logged since we passed 2 into the take operator.
takeLast
The takeLast operator returns an Observable that emits the last count values emitted by the source Observable.
It takes one argument which is the count, which the maximum number of values to emit from the end of the sequence of values.
For instance, we can use it as follows:
import { of } from "rxjs";
import { takeLast } from "rxjs/operators";
const numbers = of(1, 2, 3, 4, 5, 6).pipe(takeLast(2));
numbers.subscribe(x => console.log(x));
We should get 5 and 6 since we passed in 2 to the takeLast operator, which means we only want the last 2 values from the of(1, 2, 3, 4, 5, 6) Observable emitted.
takeUntil
The takeUntil operator returns an Observable that emits value from the source Observable until the notifier Observable emits a value.
It takes one argument, which is the notifier whose emitted value will cause the return Observable to stop emitting values from the source Observable.
For example, we can use it as follows:
import { interval, timer } from "rxjs";
import { takeUntil } from "rxjs/operators";
const source = interval(1000);
const result = source.pipe(takeUntil(timer(5000)));
result.subscribe(x => console.log(x));
The code above will emit the values from the source Observable until timer(5000) emits its first value.
So we should get something like:
0
1
2
3
takeWhile
The takeWhile operator returns an Observable that emits the values emitted by the source Observable as long as the condition in the predicate function is satisfied and completes as soon as the condition is not satisfied.
It takes of argument, which is the predicate function that returns the condition to check each value from the source Observable with. The function takes the item from the source Observable as the first parameter and the index, starting from 0, of the value emitted as the second parameter.
For example, we can use it as follows:
import { interval } from "rxjs";
import { takeWhile } from "rxjs/operators";
const source = interval(1000);
const result = source.pipe(takeWhile(x => x <= 10));
result.subscribe(x => console.log(x));
Then we should get:
0
1
2
3
4
5
6
7
8
9
10
since we specified in the predicate function that x => x <= 10, which means that any value less than or equal to 10 from the source Observable will be emitted.
The skipUntil operator returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.
skipWhile operator returns an Observable that skips all items emitted by the source Observable until the condition passed into the predicate function becomes false
The take operator returns an Observable that emits the first number of values emitted by the source Observable.
The takeLast operator returns an Observable that emits the last number of values emitted by the source Observable.
takeUntil operator returns an Observable that emits value from the source Observable until the Observable passed into this operator emits a value.
Finally, the takeWhile operator returns an Observable that emits the values emitted by the source Observable as long as the condition passed into this is satisfied.
