RxJS is a library for 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 window operators, including the windowCount, windowTime, windowToggle and windowWhen operators.
windowCount
The windowCount operator branch out the source Observable values as a nested Observable with each of them emitting at most windowSize events.
It takes up to 2 arguments. The first argument is the windowSize, which the maximum number of values to be emitted by each window.
The second argument is optional. It’s the startWindowEvery number, which defaults to 0. It’s the interval to start a new window. The interval is measured by the number of items emitted by the source Observable.
For example, we can use it as follows:
import { interval } from "rxjs";
import { windowCount, mergeAll, take, map, skip } from "rxjs/operators";
const nums = interval(1000).pipe(take(1000));
const result = nums.pipe(
windowCount(3, 3),
map(win => win.pipe(skip(1))),
mergeAll()
);
result.subscribe(x => console.log(x));
The code above has the nums Observable which emits a number every second up to 1000.
That’s pipe d to the windowCount operator which emits 3 values at a time and starts a new window after 3 values emitted from nums .
Then that is map ped to the win.pipe(skip(1)) Observable which skips 1 value for every 2 values emitted.
Finally, we pipe the values to mergeAll to merge all the Observables into one.
Then we should see that every third number isn’t emitted.
windowTime
The windowTime operator returns an Observable that emits windows of items with the window period set by the windowTimeSpan.
It takes up to 2 arguments, which is the windowTimeSpan, and the second is an optional argument, which is a scheduler object.
An example would the following:
import { interval } from "rxjs";
import { windowTime, mergeAll, take, map } from "rxjs/operators";
const nums = interval(1000);
const result = nums.pipe(
windowTime(1000, 5000),
map(win => win.pipe(take(2))),
mergeAll()
);
result.subscribe(x => console.log(x));
The code above has the nums Observable, which emits values start from 0 every second. Then the emitted values are pipe d to the windowTime operator, which starts a window every 5 seconds that’s 1 second long. Then we take 2 values from each window
This will result in 3 values being skipped in each window since values are emitted every minute by nums but we take only 2 values from every window.
windowToggle
windowToggle branches the source Observable values as nested Observable starting from emitting from openings and ending with the closingSelector emits.
It takes up to 2 arguments. The first is the openings , which is an Observable of notifications to start a new window.
The second is the closingSelector which takes the value emitted by the openings and returns an Observable which emits the next or complete signal will close the window.
We can use it as follows:
import { interval, EMPTY } from "rxjs";
import { windowToggle, mergeAll } from "rxjs/operators";
const interval$ = interval(2000);
const openings = interval(2000);
const result = interval$.pipe(
windowToggle(openings, i => (i % 3 === 0 ? interval(500) : EMPTY)),
mergeAll()
);
result.subscribe(x => console.log(x));
We have interval$ Observable which emits a number every 2 seconds. Then we have the same Observable for openings . The emitted values for interval$ are pipe d to the windowToggle operator, which has the openings Observable as the first argument, which emits every 2 seconds. So we start a new window every 2 seconds.
Then we have the second function:
i => (i % 3 === 0 ? interval(500) : EMPTY)
to close the window when the value piped in isn’t divisible by 3. This means that we get every 3 values emitted from interval$ logged.
windowWhen
The windowWhen operator branches out the source Observable using the closingSelector function, which returns an Observable to close the window.
It takes the closingSelector function which takes the value emitted by the openings and returns an Observable which emits the next or complete signal will close the window.
For example, we can use it as follows:
import { interval } from "rxjs";
import { mergeAll, take, windowWhen, map } from "rxjs/operators";
const interval$ = interval(2000);
const result = interval$.pipe(
windowWhen(() => interval(Math.random() * 4000)),
map(win => win.pipe(take(2))),
mergeAll()
);
result.subscribe(x => console.log(x));
In the code above, we have the interval$ which emits a number every 2 seconds. Then the emitted value is pipe d to the windowWhen operator, which has the closingSelector ve:
() => interval(Math.random() \* 4000)
This means the window will close and reopen Math.random() * 4000 milliseconds.
We should be that some numbers are emitted faster than others.
The windowCount operator branch out the source Observable values as nested Observable with each of them emitting at most windowSize events. windowSize is the size of each window.
windowTime operator returns an Observable that emits windows of items with the window period set by the windowTimeSpan . windowTimeSpan sets the amount of time a window is open.
windowToggle branches the source Observable values as nested Observable starting from emitting from openings and ending with the closingSelector emits. The openings and closingSelector functions are Observables that control the opening and closing of the window for each Observable respectively.
The windowWhen operator branches out the source Observable using the closingSelector function, which returns an Observable to close the window.
