RxJS Operators Explained with Example (2024)

Overview

In this post, we begin reviewing the operators, we will learn what types of operators are now included in the RxJS and which of them the RxJS team recommends to use.

Also, we will tell you about basic operators that you can typically use in your daily tasks with observables, map, filter, catch error, distant until change scan and buffer operators. So let's exploer working with operators to process stream data.

Lets understand first what operators really are.

Observable produces a sequence of data and we can subscribe to fetch it but what if we need to do some modifications to initial data, before we feed them to some component HTML template.

As RxJS implements declarative programming parity, it means that each value can be transformed with predefined function or as we name it in RxJS, operators.

RxJS Operators:

An operator is simply a method that acts on an Observable and changes the stream in some way. Operators are by nature immutable. This immutability makes the code easier to test and reason about.

Let's review definitions of operators from pre ActiveX project:

An operator is a function which creates a new observable based on the input observed. The purpose of operator, to modify or filter originally emitted values in the way we need for the project tasks.

As can be seen in the diagram, operators transfer values to an observable new output and change the values simultaneously.

There are two types of operators :

• Pipeable operators.
• Observable prototype methods.

So what is the difference and which one should we use. So one uses pipeable operators and another prototype method both of which do the same thing.

Pipeable operator is just a function so it becomes easier to combine existing operators in custom operators with specific logic. Starting from RxJS version 5.5, they are announced as preferable way of using operators.

On the other side, operators in prototype do not use Tree shaking.. if you did not know that Tree Shaking is in the process of deleting the code imported but not used in the program to reduce bundled size.

RxJS from Operator

The from() operator allows us to create an Observable from some other asynchronous/synchronous concept. It's really powerful when almost anything (array, promise, or iterable) can be made into an Observable, as this allows for rich composition.

Let's see typical example of from Operator

// RxJS v6+
    import { from } from 'rxjs';

    //emit array as a sequence of values
    let stream$ = from([1,2,3,4]);

    //output: 1,2,3,4
    stream$.subscribe( data => console.log(data)); 

RxJS Map Operator

Map operator applies a given function to each value emitted by source observable and any resulting values as an observable.

Let's see simple example of Map Operator

EXAMPLE 1: Add 10 to each number (RxJS v6+)

    import { from } from 'rxjs';
    import { map } from 'rxjs/operators';
    
    //emit (1,2,3,4)
    const source = from([1, 2, 3, 4]);
    
    //add 10 to each value
    const example = source.pipe(map(val => val + 10));
    
    //output: 11,12,13,14
    const subscribe = example.subscribe(val => console.log(val));
    
    

EXAMPLE 2: MAP Object property (RxJS v6+)

    import { from } from 'rxjs';
    import { map } from 'rxjs/operators';
    
    //emit ({name: 'Joe', age: 30}, {name: 'Frank', age: 20},{name: 'Ryan', age: 50})
    const source = from([
        { name: 'John', age: 20 },
        { name: 'Alek', age: 25 },
        { name: 'Rosy', age: 18 }
    ]);
    //grab each persons name, could also use pluck for this scenario
    const example = source.pipe(map(({ name }) => name));
    //output: "John","Alek","Rosy"
    const subscribe = example.subscribe(val => console.log(val));
    

RxJS Filter Operator

Filter operator, filters source observable items by only omitting those that satisfy a specified predicate.

Filter operator takes items emitted by the source observable and emit only those value that satisfy a specified predicate. As you know, predicate is a function that returns true or false.

Let's see simple example of filter Operator

import { from } from 'rxjs';
import { filter } from 'rxjs/operators';

let stream$ = from([1, 2]).map( x => x +1 ) .filter( x > 2 );

// output : data 3
stream$.subscribe( data => console.log('data', data)) 

Here, we can see that we are using the .map() operator and .filter() to change our stream's data. map() operates on each value in the stream by incrementing each value by one. .filter() operates on the changed stream; a change brought about by calling .map(). It also operates on each value in the stream but conditionally decides what should be emitted. The end result is only one value being emitted, 3.

RxJS distinct Operators

The main purpose of distinct operators is preventing subscribers on next handler to run, if observable emits same value again. Sometimes it makes sense since why to do view update or recalculate values if nothing is actually changed.

At the moment, RxJS has three distinct operators :

• distinctUntilChanged - Distinct until changed operator emits all items that are distinct by comparison from only one previous item, complicating compared logic is possible here.
• distinctUntilKeyChanged - Distinct until key changed works similar but allows to specify which key value you want to compare.
• distinct
- Distinct operator emits all items that are distinct by comparison from all previous items in internal buffer, you can empty its internal buffer with the second flush argument which should be an observable.

distinctUntilChanged

distinct until changed operator emits all items that are distinct by comparison (===) from only one previous item, complicating compared logic is possible here.

Distinct until changed returns an observable that emits all items emitted by the source observable that are distinct by comparison from the previous item. Pay attention that operator only compares next value with one previous value.

Let's see an example distinctUntilChanged with basic value (RxJS v6+)

    import { from } from 'rxjs';
    import { distinctUntilChanged } from 'rxjs/operators';

    // only output distinct values, based on the last emitted value
    const source$ = from([1, 1, 2, 2, 3, 3]);

    source$
        .pipe(distinctUntilChanged())
        // output: 1,2,3
        .subscribe(console.log);
    

To compare previous and next values, it uses comparison function as a first param. If a comparator function is not provided, the Equality check is used by default. If a comparator function is provided, then it will be called for each item to test for whether or not this value should be emitted.

If values are objects, you can use second pair on key selector to specify which key of the value you want to compare. But actually, the same result can be easily achieved with the justa comparison function.

distinctUntilKeyChanged

Now let's review distinct until key changed. Actually, it is very similar to distinct until changed but it has some minor differences. They have different arguments order and key name argument here is a string, so we can use only with flat objects without deep nesting. Key selector argument of distant until changed operator is a function, so we can reach object property value at any depth.

Both statements do the same thing, which one you want to use it's up to you to decide. Definitely if value is a flat object without deep nesting, distinct until key changed demands much less to type.

Example with object value (RxJS v6+)

 from([{v:1}, {v:2}, {v:2}, {v:3}])
.pipe(

    // returns all objects as, quality comparison will not work since each object has different reference. 
    distinctUntilChanged() 
    // output : {v:1}, {v:2}, {v:2}, {v:3}
    
    // with comparison callback function as the first argument for distinct until changed by using second parameter key selector
    distinctUntilChanged((prev, next) => prev.v === next.v)
    // output : {v:1}, {v:2}, {v:3}
    
    // used null as the first argument so the default strict JavaScript equality will be used.
    distinctUntilChanged(null, (item) => item.v)
    // output : {v:1}, {v:2}, {v:3}
    
    
    // very similar to distinct until changed but it has some minor differences. 
    // They have different arguments order and key name argument here is a string
    distinctUntilKeyChanged('v')
    // output : {v:1}, {v:2}, {v:3}

)

distinct

let's try distinct. As you can see, it compares with all previous values.

Let's see an example of distinct

    import { of } from 'rxjs';
    import { distinct } from 'rxjs/operators';
    of(1, 2, 3, 4, 5, 1, 2, 3, 4, 5)
        .pipe(distinct())
        // OUTPUT: 1,2,3,4,5
        .subscribe(console.log);

What is same and what is different between distinct until changed. Distinct doesn't have comparison callback so some complicated compare logic is not possible here.

It has key selector so we can specify what property value should be compared, but the main difference is next. Distinct operator compares next value, not only with one but with all previous values, so it should store values somewhere in a buffer and if number of values is significant, it can cause a lack of memory issue.

To prevent this, we have second argument here named flush. Flush should be observable, each time flush emits, all buffered values will be erased so the process starts from the beginning. For example, to empty operator internal buffer each ten seconds, we can use interval function.

As you can see, distinct compares next value with all previous values from a sequence, that's why ending value here is filtered out despite of previous item is different.

Next we will see RxJS Operators - Part II