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 transformation operators like groupBy , map , mapTo and mergeMap .
groupBy
The groupBy operator takes values emitted by the source Observable and then group them by the criteria that we set for them.
It takes 4 arguments. The first is the keySelector , which is a function that extracts the key for each item.
The second is an optional argument for the elementSelector , which is a function that extracts the return element for each item.
The 3rd argument is the durationSelector , which is an optional function that returns an Observable to determine how long each group should exist.
Finally, the last argument is the subjectSelector , which is an optional function that returns a Subject.
For example, we can use it as follows:
import { of } from "rxjs";
import { groupBy, reduce, mergeMap } from "rxjs/operators";
const observable = of(
{ id: 1, name: "John" },
{ id: 2, name: "Jane" },
{ id: 2, name: "Mary" },
{ id: 1, name: "Joe" },
{ id: 3, name: "Don" }
).pipe(
groupBy(p => p.id),
mergeMap(group$ => group$.pipe(reduce((acc, cur) => [...acc, cur], [])))
);
observable.subscribe(val => console.log(val));
In the code above, we called groupBy(p => p.id) to group the items emitted from the of Observable by id .
Then we have:
mergeMap(group$ => group$.pipe(reduce((acc, cur) => [...acc, cur], [])))
to get the grouped items together to be emitted by one Observable.
map
The map operator lets us map the values emitted by the source Observable to the other values and emits the resulting values as an Observable.
It takes up to 2 arguments. The first argument is the project function, which is required. The function takes the emitted value of the source Observable and the index of it and then returns what we want by manipulating those.
The second argument is optional. It’s the thisArg , which is used to define the this value for the project function in the first argument.
For example, we can use it as follows:
import { of } from "rxjs";
import { map } from "rxjs/operators";
const observable = of(1, 2, 3);
const newObservable = observable.pipe(map(val => val ** 2));
newObservable.subscribe(x => console.log(x));
The code above takes the emitted values from observable , then pass it to the map operator via the pipe operator. In the callback, we passed into the map operator, we exponentiate the originally emitted value to the power of 2.
Then we can subscribe to an Observable that emits the new values and we get:
1
4
9
logged.
mapTo
The mapTo operator emits the given constant value for any source Observable’s emitted value.
It takes one argument, which is the value to emit.
For example, we can use it as follows:
import { of } from "rxjs";
import { mapTo } from "rxjs/operators";
const observable = of(1, 2, 3);
const newObservable = observable.pipe(mapTo("foo"));
newObservable.subscribe(x => console.log(x));
The code above maps all the values from observable to the value 'foo' , so we get 'foo' 3 times instead of 1, 2 and 3.
mergeMap
The mergeMap operator takes the values emitted from a source Observable and then lets us combine it with the values of another Observable.
It takes up to 3 arguments. The first is a project function to project the values and return a new Observable.
The second argument is an optional argument that takes an resultSelector . We can pass in a function to select the values to emit from the result.
The third argument is the concurrency , which is an optional argument that specifies the maximum number of input Observables to be subscribed to concurrently. The default is Number.POSITIVE_INFINITY .
For example, we can use it as follows:
import { of } from "rxjs";
import { mergeMap, map } from "rxjs/operators";
const nums = of(1, 2, 3);
const result = nums.pipe(mergeMap(x => of(4, 5, 6).pipe(map(i => x + i))));
result.subscribe(x => console.log(x));
The code above will get the values from the 3 values emitted from the nums Observable, then pass the values to the mergeMap ‘s callback function via the pipe operator. x will have the values from nums .
Then in the callback, we have the of(4, 5, 6) Observable, which have the values from combined from the nums Observable. i has the values from the of(4, 5, 6) Observable, so we the values from both Observables added together. We get 1 + 4, 1 + 5, 1 + 6, 2 + 4, 2 + 5, 2 + 6 and so on.
In the end, we should get the following output:
5
6
7
6
7
8
7
8
9
The groupBy operator takes values emitted by the source Observable and then group them by the criteria that we set for them. We can use it in conjunction with the mergeMap to combined the results grouped by the groupBy operator into one Observable.
The map operator lets us map the values emitted by the source Observable to the other values and emits the resulting values as an Observable.
The mapTo operator emits the given constant value for any source Observable’s emitted value.
Finally mergeMap operator takes the values emitted from a source Observable and then lets us combine it with the values of another Observable. Then we get an Observable with the values of both Observables combined together.
