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 some filter operators, including elementsAt , filter , first , ignoreElements , and last .

elementAt

The elementAt operator lets us get a single value by at the specified index in a sequence of emissions from the source Observable.

It takes up to 2 arguments. The first is the index , which is the number starting from 0 which the source Observable has emitted since subscription.

The second is the defaultValue which is an optional argument for the default value returned if the value is with the given index is not found.

It returns an Observable that emits a single item if it’s found. Otherwise, the default value provided by the optional second argument will be emitted.

A ArgumentOutOfRangeError will be thrown if the index provided is less than 0 or the Observable has completed before emitting the emission with the givenindex .

For example, we can use it as follows:

import { of } from "rxjs";  
import { elementAt } from "rxjs/operators";

const of$ = of(1, 2, 3);  
const result = of$.pipe(elementAt(2));  
result.subscribe(x => console.log(x));

Then we get 3 from console.log .

filter

The filter operator lets us filter out the emitted values from the source Observable by emitting the values from the source Observable that satisfies the conditions that are returned by a predicate function.

It takes up to 2 arguments. The first is the predicate function that’s evaluated each time the source Observable emits a value. The value is checked against this and it’s emitted by the returned Observe if it meets the criteria.

The predicate function takes 2 parameters, which are the object emitted from the source Observable and the index respectively. index starts from 0 and it’s the index -th object emitted since subscription.

For example, we can use it as follows:

import { of } from "rxjs";  
import { filter } from "rxjs/operators";

const of$ = of(1, 2, 3, 4, 5, 6);  
const result = of$.pipe(filter(x => x % 2 === 0));  
result.subscribe(x => console.log(x));

Then we get:

2  
4  
6

first

The first operator emits only the first value emitted by the source Observable.

It takes 2 optional arguments. The first is the predicate function which is called with each item to test for conditional matching.

The second is the default value, which is optional. It’s the default value that’s emitted in case no valid value was found from the source.

For example, we can use it as follows:

import { of } from "rxjs";  
import { first } from "rxjs/operators";

const of$ = of(1, 2, 3, 4, 5, 6);  
const result = of$.pipe(first());  
result.subscribe(x => console.log(x));

The code above takes the values from the of$ Observable and emits the first one, so we get 1 from the console.log .

We can also pass in a function to check for a match from the source Observable. For example, we can write:

import { of } from "rxjs";  
import { first } from "rxjs/operators";

const of$ = of(1, 2, 3, 4, 5, 6);  
const result = of$.pipe(first(x => x % 3 === 0));  
result.subscribe(x => console.log(x));

Since we passed in x => x % 3 === 0 to thefirst operator, first will check for the first one that matches our condition, which is a number that’s evenly divisible by 3.

Then we get the value 3 from console.log since that’s the first number of of$ that matches the condition.

We can pass in a default value as the second argument of first :

import { of } from "rxjs";  
import { first } from "rxjs/operators";  
const of$ = of(1, 2, 3, 4, 5, 6);  
const result = of$.pipe(first(x => x % 3 === 10, "none"));  
result.subscribe(x => console.log(x));

Since there’s no number in of$ that has the remainder 10 when divided by 3, the 'none' string will be emitted by the returned Observable.

ignoreElements

ignoreElements ignores all items emitted by the source Observable and only passes calls of complete or error .

It takes no arguments.

For example, we can use it as follows:

import { of } from "rxjs";  
import { ignoreElements } from "rxjs/operators";  
const of$ = of(1, 2, 3, 4, 5, 6);  
const result = of$.pipe(ignoreElements());  
result.subscribe(  
  val => console.log(val),  
  err => console.log(err),  
  () => console.log("end")  
);

Since nothing from the of$ Observable is emitted before completion, we just get 'end' logged.

last

The last operator returns an Observable that emits only the last item that’s emitted by the source Observable. If a predicate function is passed into last , then the last value that matches the condition returned by the predicate will be emitted.

It takes 2 optional arguments. The first is the predicate, which is optional. The predicate returns the condition that the emitted value from the source has to satisfy.

The second is an optional argument for the defaultValue , which the value that’ll be emitted if nothing emitted from the source Observable meets the condition returned by the predicate function.

For example, we can use it as follows:

import { of } from "rxjs";  
import { last } from "rxjs/operators";  
const of$ = of(1, 2, 3, 4, 5, 6);  
const result = of$.pipe(last());  
result.subscribe(val => console.log(val));

Then we get 6 logged since it’s the last value emitted by the of$ Observable.

We can also specify a condition as follows:

import { of } from "rxjs";  
import { last } from "rxjs/operators";  
const of$ = of(1, 2, 3, 4, 5, 6);  
const result = of$.pipe(last(x => x % 2 === 1));  
result.subscribe(val => console.log(val));

Then we get 5 logged since it’s the last value from the of$ Observable that’s odd.

elementAt operator lets us get a single value by at the specified index in a sequence of emissions from the source Observable.

The filter operator only emits the values from the source Observable that satisfies the conditions that are returned by a predicate function.

first operator emits only the first value emitted by the source Observable.

ignoreElements and only passes calls of complete or error and ignores everything else.

last returns an Observable that emits only the last item that’s emitted by the source Observable or the last one that meets the condition in the predicate function if it’s specified.

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 some filtering operators, including the distinct , distinctUntilChanged , and distinctUntilKeyChanged operators.

distinct

The distinct operator emits the items from the source Observable that are distinct in comparison to previous items from the source.

It takes 2 optional arguments. The first is the keySelector function, which lets us select which value we want to check as distinct.

The second is an optional flushes Observable for flushing the internal HashSet from the operator.

It returns a new Observable that emits values that are distinct.

A simple example would be the following:

import { of } from "rxjs";  
import { distinct } from "rxjs/operators";
of(3, 3, 3, 3, 3, 3, 35, 5, 7, 8, 4, 6, 3, 5, 2, 4, 2)  
  .pipe(distinct())  
  .subscribe(x => console.log(x));

We have the following Observable with duplicate values:

of(3, 3, 3, 3, 3, 3, 35, 5, 7, 8, 4, 6, 3, 5, 2, 4, 2)

The values from it is pipe d to the distinct() operator to filter out duplicate values.

Then we get:

3  
35  
5  
7  
8  
4  
6  
2

from the console.log .

We can also check if some key of an entry is distinct as follows:

import { of } from "rxjs";  
import { distinct } from "rxjs/operators";

const people = [  
  { age: 4, name: "Joe" },  
  { age: 7, name: "Jane" },  
  { age: 5, name: "Jane" }  
];

of(...people)  
  .pipe(distinct(p => p.name))  
  .subscribe(x => console.log(x));

The code above spreads the people array as the arguments of the of operator, which emits the objects in the people array. Then the emitted values are pipe d into the distinct operator, which selects the name property to check for distinctness.

Then only the objects with a different name value is emitted.

In the end, we get:

{age: 4, name: "Joe"}  
{age: 7, name: "Jane"}

As the Observable.

distinctUntilChanged

distinctUntilChanged emits all items emitted from the source Observable that are distinct by comparison from the previous item from the source.

It takes 2 optional arguments, which is a compare function to test if an item is distinct from the previous item. The second argument is the keySelector function to return the value of the key that we want to check.

A simple example would be as follows:

import { of } from "rxjs";  
import { distinctUntilChanged } from "rxjs/operators";
of(1, 1, 5, 5, 6, 7, 8, 8, 8, 8, 9)  
  .pipe(distinctUntilChanged())  
  .subscribe(x => console.log(x));

The values from the of(1, 1, 5, 5, 6, 7, 8, 8, 8, 8, 9) is pipe d to the distinctUntilChanged operator and the previously emitted value is checked against the currently emitted value to see if they’re the same.

Then we get:

1  
5  
6  
7  
8  
9

since the value that’s different from the previously emitted one is emitted.

For Observables that emit objects, we can check if the property’s value is considered the same by passing in a function to the distinctUntilChanged operator as follows:

import { of } from "rxjs";  
import { distinctUntilChanged } from "rxjs/operators";  
const people = [  
  { age: 4, name: "Joe" },  
  { age: 7, name: "Jane" },  
  { age: 5, name: "Jane" }  
];  
of(...people)  
  .pipe(distinctUntilChanged((p, q) => p.name === q.name))  
  .subscribe(x => console.log(x));

The first and last parts work like the previous example. The difference is that now we have the (p, q) => p.name === q.name function to check if the previously emitted name value of the emitted object is the same as the currently emitted one.

Then we get:

{age: 4, name: "Joe"}  
{age: 7, name: "Jane"}

as the output from console.log .

distinctUntilKeyChanged

distinctUntilKeyChanged returns an Observable that emits the value of a source Observable that’s distinct by the comparison with a key of the object emitted previously from the source.

This means that an item will be emitted if the value with the given key is different from the previously emitted object key’s value.

It takes up to 2 arguments. The first is the key , which is a string with the object property to look up for each item.

The second is an optional compare function, which is called to test if an item is distinct from the previously emitted item from the source Observable.

It returns an Observable that emits item from the source Observable with distinct property value from the previously emitted one from the source.

For example, we can rewrite the previous example:

import { of } from "rxjs";  
import { distinctUntilChanged } from "rxjs/operators";  
const people = [  
  { age: 4, name: "Joe" },  
  { age: 7, name: "Jane" },  
  { age: 5, name: "Jane" }  
];  
of(...people)  
  .pipe(distinctUntilChanged((p, q) => p.name === q.name))  
  .subscribe(x => console.log(x));

into:

import { of } from "rxjs";  
import { distinctUntilKeyChanged } from "rxjs/operators";  
const people = [  
  { age: 4, name: "Joe" },  
  { age: 7, name: "Jane" },  
  { age: 5, name: "Jane" }  
];  
of(...people)  
  .pipe(distinctUntilKeyChanged("name"))  
  .subscribe(x => console.log(x));

Then we should get the same result as before. All we did was changing:

distinctUntilChanged((p, q) => p.name === q.name)

to:

distinctUntilKeyChanged("name")

We can use the distinct operator to get the distinct values emitted from a source Observable either by comparing primitive values or values of the properties of an object.

distinctUntilChanged , and distinctUntilKeyChanged let us emit the items from a source Observable that are different from the ones previously emitted by the source Observable. Again, it can compare by the primitive values or the values of the properties of an object.