Categories
JavaScript Rxjs

Rxjs Operators — Calculation and Aggregation

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.

Categories
JavaScript Rxjs

Rxjs Filtering Operators — Skip and Take

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 more filtering operators, including skipUntil, skipWhile, take, takeLast, takeUntil, and takeWhile operators.

skipUntil

The skipUntil operator returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.

It takes one argument, which is the notifer Observable. When this emits an item, then the source Observable’s emitted values will be emitted by the skipUntil operator.

For example, we can use it as follows:

import { interval } from "rxjs";  
import { skipUntil, take } from "rxjs/operators";

const numbers = interval(1000).pipe(  
  take(10),  
  skipUntil(interval(5000).pipe(take(1)))  
);  
numbers.subscribe(x => console.log(x));

The code above takes the emitted values from the interval(1000) Observabkem then pipe it to the take(10) operator to take the first 10 values, then take those values to the skipUntil operator, which has the interval(5000).pipe(take(1)) Observable passed in.

interval(5000).pipe(take(1)) emits a value after 5 seconds, which then will trigger the emission of the interval(1000) Observable’s emitted values.

Then we should see:

4  
5  
6  
7  
8  
9

skipWhile

The skipWhile operator returns an Observable that skips all items emitted by the source Observable until the condition returned by the predicate function becomes false.

It takes one argument, which is a predicate function to test each item with the condition returned by the function.

For example, we can write the following:

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

const numbers = of(1, 2, 3, 4, 5, 6).pipe(skipWhile(x => x < 3));  
numbers.subscribe(x => console.log(x));

Then the items from the of(1, 2, 3, 4, 5, 6) Observable won’t be emitted when the value is less than 3. This means that numbers will emit 3, 4, 5 and 6.

Once the condition in the predicate function is tested false once, it’ll keep emitting regardless whether the emitted value from the source Observable make the condition evaluate to false or not.

So if we have:

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

const numbers = of(1, 2, 3, 4, 3, 2).pipe(skipWhile(x => x < 3));  
numbers.subscribe(x => console.log(x));

Then we get 3, 4, 5 and 2 since the condition becomes false once the first 3 was emitted by of(1, 2, 3, 4, 3, 2).

take

The take operator returns an Observable that emits the first count values emitted by the source Observable.

It takes one argument which is the count, which the maximum number of values to emit.

For example, we can use it as follows:

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

const numbers = of(1, 2, 3, 4, 3, 2).pipe(take(2));  
numbers.subscribe(x => console.log(x));

Then we get 1 and 2 logged since we passed 2 into the take operator.

takeLast

The takeLast operator returns an Observable that emits the last count values emitted by the source Observable.

It takes one argument which is the count, which the maximum number of values to emit from the end of the sequence of values.

For instance, we can use it as follows:

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

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

We should get 5 and 6 since we passed in 2 to the takeLast operator, which means we only want the last 2 values from the of(1, 2, 3, 4, 5, 6) Observable emitted.

takeUntil

The takeUntil operator returns an Observable that emits value from the source Observable until the notifier Observable emits a value.

It takes one argument, which is the notifier whose emitted value will cause the return Observable to stop emitting values from the source Observable.

For example, we can use it as follows:

import { interval, timer } from "rxjs";  
import { takeUntil } from "rxjs/operators";

const source = interval(1000);  
const result = source.pipe(takeUntil(timer(5000)));  
result.subscribe(x => console.log(x));

The code above will emit the values from the source Observable until timer(5000) emits its first value.

So we should get something like:

0  
1  
2  
3

takeWhile

The takeWhile operator returns an Observable that emits the values emitted by the source Observable as long as the condition in the predicate function is satisfied and completes as soon as the condition is not satisfied.

It takes of argument, which is the predicate function that returns the condition to check each value from the source Observable with. The function takes the item from the source Observable as the first parameter and the index, starting from 0, of the value emitted as the second parameter.

For example, we can use it as follows:

import { interval } from "rxjs";  
import { takeWhile } from "rxjs/operators";

const source = interval(1000);  
const result = source.pipe(takeWhile(x => x <= 10));  
result.subscribe(x => console.log(x));

Then we should get:

0  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10

since we specified in the predicate function that x => x <= 10, which means that any value less than or equal to 10 from the source Observable will be emitted.

The skipUntil operator returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.

skipWhile operator returns an Observable that skips all items emitted by the source Observable until the condition passed into the predicate function becomes false

The take operator returns an Observable that emits the first number of values emitted by the source Observable.

The takeLast operator returns an Observable that emits the last number of values emitted by the source Observable.

takeUntil operator returns an Observable that emits value from the source Observable until the Observable passed into this operator emits a value.

Finally, the takeWhile operator returns an Observable that emits the values emitted by the source Observable as long as the condition passed into this is satisfied.

Categories
JavaScript Rxjs

Rxjs Filtering Operators — Sample, Skip and Single

RxJS is a library for doing reactive programming. Creating 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 sample, sampleTime,skip, single, and skipLast operators.

sample

The sample operator returns an Observable that emits the value from the source Observable when the notifier Observable emits.

It takes one argument, which is the notifer Observable.

For example, we can use it as follows:

import { interval } from "rxjs";  
import { sample } from "rxjs/operators";
const seconds = interval(1000);  
const result = seconds.pipe(sample(interval(5000)));  
result.subscribe(x => console.log(x));

In the code above, we have the seconds Observable which emits a number every second. Then we pipe the result of it to the sample operator, where we set the notifier Observable which is interval(5000), which emits every 5 seconds.

Therefore, we have a new Observable that emits values from seconds every 5 seconds.

We should see every few values from seconds logged.

sampleTime

The sampleTime operator emits the most recently emitted value from the source Observable in periodic intervals.

It takes up to 2 arguments. The first is the period, which is the period to wait until emitting a value from the source Observable. It’s measured in milliseconds or the time unit of the optional scheduler .

The second argument is the optional scheduler, which defaults to async.

It returns a new Observable that emits value from the source Observable in a specified time interval.

For example, we can use it as follows:

import { interval } from "rxjs";  
import { sampleTime } from "rxjs/operators";
const seconds = interval(1000);  
const result = seconds.pipe(sampleTime(5000));  
result.subscribe(x => console.log(x));

The code above takes the values emitted from the seconds Observable and pipe it to the sampleTime operator, which will emit values from seconds every 5 seconds.

The result will be that we get values emitted from the result Observable, which gets its values from the seconds Observable, every 5 seconds.

We should see some numbers logged from the console.log .

single

single returns an Observable that emits the single item from the source Observable that matches a given condition returned by the predicate function.

If the source Observable emits more than one of such item or no items, then we get IllegalArgumentException or NoSuchElementException respectively.

If the source Observable emits items but none match the condition in the predicate function then undefined is emitted.

It takes one argument, which is the predicate function that returns the condition to match the item emitted from the source Observable.

For example, if we have:

import { of } from "rxjs";  
import { single } from "rxjs/operators";
const numbers = of(1).pipe(single());  
numbers.subscribe(x => console.log(x));

Then we get 1 logged.

If we have:

import { of } from "rxjs";  
import { single } from "rxjs/operators";
const numbers = of(1, 2).pipe(single());  
numbers.subscribe(x => console.log(x));

Then we get the error “Sequence contains more than one element” logged.

If we have:

import { of } from "rxjs";  
import { single } from "rxjs/operators";
const numbers = of(1, 2).pipe(single(x => x % 2 === 0));  
numbers.subscribe(x => console.log(x));

Then we get 2 logged since 2 matches the condition x => x % 2 === 0 .

skip

The skip operator returns an Observable that skips the first count items emitted by the source Observable and emit the rest.

It takes the required count argument, which is the number of items from the source Observable to skip.

For example, if we have:

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

const numbers = of(1, 2, 3, 4, 5).pipe(skip(2));  
numbers.subscribe(x => console.log(x));

Then we get:

3  
4  
5

logged because we specified that we want to skip the first 2 emitted values from of(1, 2, 3, 4, 5) .

skipLast

The skipLast operator returns an Observable that skips the last count values emitted from the source Observable.

It takes the required count argument, which is the number of items from the source Observable to skip from the end of the source Observable.

For example, we can write:

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

const numbers = of(1, 2, 3, 4, 5).pipe(skipLast(2));  
numbers.subscribe(x => console.log(x));

Then we get:

1  
2  
3

logged since we specified that we want to skip the last 2 emitted values.

The sample and sampleTime operators return an Observable that emits the value from the source Observable when the notifier Observable emits or at the specified time interval respectively.

single returns an Observable that emits the single item from the source Observable that matches a given condition.

The skip operator returns an Observable that skips the number of items at the start of emitting the values from the Observable.

Finally, skipLast operator returns an Observable that skips the last number of values emitted from the source Observable.

Categories
JavaScript Rxjs

Rxjs Filtering Operators — Getting Specific Values

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.

Categories
JavaScript Rxjs

Rxjs Filtering Operators — Distinctness

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.