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 calculation and aggregation operators including count , max , min and reduce .

count

The count operator returns an Observable that counts the number of emissions on the source and emits the number when the source completes.

It takes an optional predicate function which returns a boolean with the condition for counting.

The predicate has 3 parameters, which is the value emitted by the source Observable, the index , which is the zero-based ‘index’ of the value from the source Observable, and the source , which is the source Observable itself.

For example, we can use it as follows to count the total number of emitted values:

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

The count operator without a predicate function passed in will count all emissions from the source Observable.

This should get us 6 since we have 6 values in the of$ Observable.

We can also pass in a predicate function as follows:

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

Then we should get 3 since we’re only the number of even numbers emitted from of$ .

max

The max operator takes value from a source Observable that emits the numbers or items that can be compared with a comparer function, and gets the largest one from it and emits it with the returned Observable.

It takes an optional comparer function, which we can use to compare the value of 2 items.

For example, we can use it to get the highest number emitted from an Observable as follows:

import { of } from "rxjs";  
import { max } from "rxjs/operators";
of(2, 3, 4, 100)  
  .pipe(max())  
  .subscribe(x => console.log(x));

We should get 100 since 100 is the biggest number in the of(2, 3, 4, 100) Observable.

Also, we can pass in a function to the max operator to get the largest value from a list of objects as follows:

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

const people = [  
  { age: 17, name: "Joe" },  
  { age: 25, name: "Jane" },  
  { age: 19, name: "Mary" }  
];

of(...people)  
  .pipe(max((a, b) => a.age - b.age))  
  .subscribe(x => console.log(x));

Then we should get:

{age: 25, name: "Jane"}

from the console.log .

The comparer function works like the callback of the array’s sort method. We keep the order if comparer returns a negative number. We switch the order if comparer returns a positive number. Otherwise, they’re the same. Then the largest one is picked from the end.

min

The min operator is the opposite of the max operator. It gets the smallest value from the source Observable.

The arguments are the same as the max operator.

We can use it like the max operator as follows:

import { of } from "rxjs";  
import { min } from "rxjs/operators";
of(2, 3, 4, 100)  
  .pipe(min())  
  .subscribe(x => console.log(x));

Then we get 2.

Like the max operator, we can use the min operator with a comparer function as follows:

import { of } from "rxjs";  
import { min } from "rxjs/operators";  
const people = [  
  { age: 17, name: "Joe" },  
  { age: 25, name: "Jane" },  
  { age: 19, name: "Mary" }  
];  
of(...people)  
  .pipe(min((a, b) => a.age - b.age))  
  .subscribe(x => console.log(x));

Then we get:

{age: 17, name: "Joe"}

The comparer function works the same way as the one we pass into the max operator, except that the first one is picked instead of the last.

reduce

The reduce operator applies an accumulator function over all the values emitted by the source Observable to combine them into a single value, which is emitted by the returned Observable.

It takes up to 2 arguments. The first is the accumulator function, which is the function we use to combine the values.

The second argument is an optional seed value, which is the initial value of the accumulation.

For example, we can use it as follows:

import { of } from "rxjs";  
import { reduce } from "rxjs/operators";  
of(2, 3, 4, 100)  
  .pipe(reduce((a, b) => a + b, 0))  
  .subscribe(x => console.log(x));

The code above will sum up of the values from the of(2, 3, 4, 100) Observable and emits 109, which are the sum of 2, 3, 4, and 100.

a and b are the values emitted from the source Observable.

The count operator returns an Observable to count the number of times a source Observable emits or source Observable emitting a certain kind of value depending on the condition we specify.

min and max are used to get the minimum and maximum value according to a sorting function or from a group of numbers emitted respectively.

The reduce operator returns an Observable that takes the values from the source Observable and combine them into one in a way that we specify in the function we pass into it.