RxJS Operators in Angular

RxJS operators are special types of function that act on observables, taking one or more observable as input and returning a new observable with modified behavior by doing various kind of operation. We do operation before subscribing and these operations are applied to the data before it is subscribed to.

There is many operators available in RxJS. It is categorized in two types:

1. Creation Operators
2. Pipeable operators

Creation operators are used to create new observables,
of, from, interval and timer are few examples of creation operators.

Pipeable operators are used to transform, filter or modify data emitted by an observable. They take an observable as input, process the emitted values and return a new observable.

pipeable operator are used within the ‘pipe’ method of an observable.

map, filter, mergeMap, debounceTime, combineLatest, first , take are the few examples of pipeable operators.

Lets understand and see some examples of above listed operators.

1. of
   The ‘of’ operator in RxJS is a creation operator used to create an observable that emits a sequence of values that you provide as arguments. It allows you to quickly create an observable with predefined set of values.
   It usually used for testing or creating example data of observable.

observable= of(1,2,3,4);
observable1= of([1,2,3,4]);

this.observable.subscribe((data)=>{
      console.log(data)
})
// 1 2 3 4

this.observable1.subscribe((data)=>{
      console.log(data)
 })
 // [1,2,3,4]

2. from
This operator is used to create observable from various types of input data.

• It is used to create observables from arrays, iterables or other iterable-like-objects.
• It is used to create observables from a promise.

  numbers = from([1, 2, 3]);
  numbers1 = from([
    { name: 'bittu', city: 'blr' },
    { name: 'bittu', city: 'blr' },
  ]);
  numbers2 = from('hello');
  numbers3 = from(new Promise((resolve, reject) => resolve('hello')));
  numbers4 = from(fetch('https://jsonplaceholder.typicode.com/users/1'));

 
*************
    

    this.numbers.subscribe((data) => {
      console.log('from', data);
    });

    this.numbers1.subscribe((data) => {
      console.log('from', data);
    });

    this.numbers2.subscribe((data) => {
      console.log(data);
    });

    this.numbers3.subscribe((data) => {
      console.log(data);
    });

    this.numbers4.subscribe((data) => {
      let numbers5 = from(data.json());
      numbers5.subscribe((data) => {
        console.log(data);
      });
    });

output from from operator example

3. pipe
The pipe operator is powerful tool in RxJS and It is used to take multiple operator as input and do some operation then return a new modified observable.
If we want to do multiple operation on observable using multiple operator so in that case we use pipe operator and inside pipe we pass all required operators.

  numbers = from([1, 2, 3, 4]);   


  this.numbers
      .pipe(
        map(function (x) {
          return x * 2;
        }),
        filter(function(x){
          return x>2
        })
      )
      .subscribe((data) => {
        console.log("pipe- map, filter operators",data);
      });

output: pipe operator- map and filter

4. map
The ‘map’ operator in RxJS is a transformation operator used to transform each value emitted by an observable into a new value. It allows you to apply a function to each item emitted by the source observable and produce a new observable that emits the transformed values.

 
    observable = of(1, 2, 3, 4);
    observable1 = of([1, 2, 3, 4]);


      this.observable.
      pipe(
        map(function (value){
          return value*2
        })
      ).
      subscribe((data)=>{
        console.log("map in number",data)
      })


      this.observable1.
      pipe(
        map(function (array){
          return array.map(value=>value*2)
        })
      ).
      subscribe((data)=>{
        console.log("map in array",data)
      })

output- map operator

5. filter
The ‘filter’ operator is used to selectively emit values from an observable based on a specified condition.

  observable = of(1, 2, 3, 4);
  numbers = from([1, 2, 3, 4]);
  

  this.observable.pipe(filter((value) => value > 2)).subscribe((data) => {
      console.log('filter data in number', data);
    });

    this.numbers.pipe(filter((value) => value > 2)).subscribe((data) => {
      console.log('filter data in numberbbb', data);
    });

outout- filter operator

6. first

 
    observable = of(1, 2, 3, 4);
    observable1 = of([1, 2, 3, 4]);

    this.observable.pipe(first()).subscribe((data) => {
      console.log('first operator', data);
    });

    this.observable1.pipe(first()).subscribe((data) => {
      console.log('first operator', data);
    });

output- first operator

7. interval
This operator create an observable that emits sequential value every specified interval of time. It is commonly used when need to produce a continuous stream of value at regular interval.

It is important to unsubscribe from the observable whenever you no longer need it to prevent from memory leakage.

source = interval(1000);    
subscribe: Subscription;

this.subscribe = this.source.subscribe(val => console.log(val));

ngOnDestroy() {
    this.subscribe.unsubscribe();
}

8. take
This is used when need to emits only the first count values emitted by the source Observable. In our below example It will emit only first 4 value.

  source = interval(1000); 
  takeFourNumber: Subscription;

  this.takeFourNumber= this.source.pipe(take(4)).subscribe((data)=>{
      console.log(data)
  })

9. forkJoin

• We use forkJoin to combine these observables into a single observable.
• In the subscription, we receive the combined results as an array.
• forkJoin is RxJS operator that allows you to combine the result of multiple observables and emits those result as an array with all the source observable (requested)
• It is a powerful tool for performing multiple asynchronous operation in parallel and waiting for all of them to complete before taking further action.

  response=[]

  ngOnInit(){
    const request1 = this.http.get('https://jsonplaceholder.typicode.com/users/1');
    const request2 = this.http.get('https://jsonplaceholder.typicode.com/users/2');
    const request3 = this.http.get('https://jsonplaceholder.typicode.com/users/3');
    const request4 = this.http.get('https://jsonplaceholder.typicode.com/users/4');

    forkJoin([request1, request2, request3, request4]).subscribe(
      (data) => {
        this.response = data;
        console.log(this.response)
      },
      (error) => {
        console.error('Error:', error);
      }
    );

    forkJoin([request1, request2, request3]).subscribe(
      ([result1, result2, result3]) => {
        console.log('Result 1:', result1);
        console.log('Result 2:', result2);
        console.log('Result 3:', result3);
      },
      (error) => {
        console.error('Error:', error);
      }
    )

forkJoin Example output

10. combineLatest

• It is RxJS operator that combines multiple observables into a single observable that emits an array of the latest values from each of the input observables whenever any of the input observables emit a value.
• This is similar to forkJoin, in case of forkJoin- emits result as observable when all observable finishes but in case of combineLatest, emits result the latest combined value from each observable.
• combineLatest will not emit an initial value until each observable emits at least one value.

  //combine latest
    //timerOne emits first value at 6s, then once every 4s
    const timer1 = timer(6000, 4000).pipe(take(4));
    //timerTwo emits first value at 5s, then once every 4s
    const timer2 = timer(5000, 4000).pipe(take(5));
    //timerThree emits first value at 7s, then once every 4s
    const timer3 = timer(7000, 4000).pipe(take(3));

    //when one timer emits, emit the latest values from each timer as an array
    combineLatest([timer1, timer2, timer3]).subscribe(
      ([timerValOne, timerValTwo, timerValThree]) => {
        this.combinedLatestSingleValues = [
          timerValOne,
          timerValTwo,
          timerValThree,
        ];
        console.log('latest value', this.combinedLatestSingleValues);
      }
    );

output- combineLatest RxJS Operator

11. concat

• The concat operator in RxJS combines multiple observables in a specific order.
• It ensures that observables are concatenated sequentially, one after the other.
• This is helpful when you want to maintain a specific order of execution for asynchronous operations, such as making HTTP requests in a specific sequence.

    const timer12 = interval(1000).pipe(take(10));
    const timer13 = interval(2000).pipe(take(6));
    const timer14 = interval(500).pipe(take(10));

    const result = concat(timer12, timer13, timer14);
    result.subscribe((x) => {
      console.log('concat', x);
    });



    const sourceA$ = of(1, 2, 3);
    const sourceB$ = of(4, 5, 6);
    const sourceC$ = of(7, 8, 9);
    const source$ = concat(sourceA$, sourceB$, sourceC$);

    source$.subscribe((data) => console.log(data));

output: concat RxJS operator

12. catchError

13. retry

14. mergeMap

For more detailed information, You can refer his official RxJS page

Thanks for reading.

Happy learning, see you soon in next exciting JavaScript or Angular article.