RxJS

Listed frequently used operators for ease of understanding

Please check on depreciations from official website reference below.

Combination

• combineLatest — To combine multiple observables into one observable and emit value as one object with latest/last values from all observables.

Initially, it emits only when each observable emits atleast one value. After inital values, it will emit every time any of the observables emits value. Everytime it emits, it will emit current values of all observables.

// RxJS v6+
import { timer, combineLatest } from 'rxjs';

// timer emits value as 0,1,2,3,4 ...
// timerOne emits first value at 1s, then once every 4s
const timerOne$ = timer(1000, 4000);
const timerTwo$ = timer(2000, 4000);
const timerThree$ = timer(3000, 4000);

// when one timer emits, emit the latest values from each timer as an array
combineLatest(timerOne$, timerTwo$, timerThree$).subscribe(
  ([timerValOne, timerValTwo, timerValThree]) => {
    /*
   Example:
    timerThree first tick: 'Timer One Latest: 0, Timer Two Latest: 0, Timer Three Latest: 0
    timerOne second tick: 'Timer One Latest: 1, Timer Two Latest: 0, Timer Three Latest: 0
    timerTwo second tick: 'Timer One Latest: 1, Timer Two Latest: 1, Timer Three Latest: 0
  */
  }
);

• concat — To combine multiple observables into single observable but emit values in order which they were added and only if previous subscription is complete go to next in order.

Scenario:- When loading images, load next image only when previous image is downloaded completely.

// RxJS v6+
import { of, concat } from 'rxjs';

concat(
  of(1, 2, 3),
  // subscribed after first completes
  of(4, 5, 6),
  // subscribed after second completes
  of(7, 8, 9)
)
  // log: 1, 2, 3, 4, 5, 6, 7, 8, 9
  .subscribe(console.log);
-----------------------------------------------------------------------

// RxJS v6+
import { interval, of, concat } from 'rxjs';

// when source never completes, any subsequent observables never run
concat(interval(1000), of('This', 'Never', 'Runs'))
  // log: 1,2,3,4.....
  .subscribe(console.log);
-----------------------------------------------------------------------

import { concat, interval } from 'rxjs';
import { take } from 'rxjs/operators';

const timer1 = interval(1000).pipe(take(1)); // 0
const timer2 = interval(2000).pipe(take(2)); // 0, 1

const output = concat(timer1, timer2);

output.subscribe(value => console.log(value)); 
// take will take first n number of emissions given and completes subscription

• forkJoin — When you have a group of observables and only care about the final emitted value of each of observable but not worried about previous values that each observable has emitted at that point of time.

// RxJS v6+
import { delay, catchError } from 'rxjs/operators';
import { forkJoin, of, throwError } from 'rxjs';

/*
  Emit values from successfully completed
  inner observables.
*/
const example = forkJoin({
  // emit 'Hello' immediately
  sourceOne: of('Hello'),
  // emit 'World' after 1 second
  sourceTwo: of('World').pipe(delay(1000)),
  // throw error
  sourceThree: throwError('This will error').pipe(catchError(error => of(error)))
});

/*
 * Output:
 * {
 *   sourceOne: "Hello",
 *   sourceTwo: "World",
 *   sourceThree: "This will error"
 * }
 */
const subscribe = example.subscribe(val => console.log(val));

------------------------------------------------------------------------------

import { forkJoin, of, timer } from 'rxjs';

const observable = forkJoin({
  foo: of(1, 2, 3, 4), // An observable emitting an array of values
  bar: Promise.resolve(8), // A promise resolving to a single value
  baz: timer(3000), // An observable that emits a value after 3 seconds
  hek: [of(1, 2, 3), 5], // An array of an observable and a single value
});

observable.subscribe({
  next: (value) => console.log(value),
  complete: () => console.log('This is how it ends!'),
});

// Outputs
// { foo:4, bar:8, baz:0, hek: 5 }

• merge — To combine multiple observables which produce stream of data to one singe output stream of data. Data can come in any order and it will start emitting data as soon as atleast one observable emits data. Unlike combineLatest which waits for all observables to emit data atleast once.

Scenario:- When messages come from multiple sources in chat, we need to merge and and show them in received order.

// From rxjs website
import { mapTo } from 'rxjs/operators';
import { interval, merge } from 'rxjs';
//emit every 2.5 seconds
const first = interval(2500);
//emit every 2 seconds
const second = interval(2000);
//emit every 1.5 seconds
const third = interval(1500);
//emit every 1 second
const fourth = interval(1000);
//emit outputs from one observable
const example = merge(
  first.pipe(mapTo('FIRST!')),
  second.pipe(mapTo('SECOND!')),
  third.pipe(mapTo('THIRD')),
  fourth.pipe(mapTo('FOURTH'))
);
//output: "FOURTH", "THIRD", "SECOND!", "FOURTH", "FIRST!", "THIRD", "FOURTH"
const subscribe = example.subscribe(val => console.log(val));

Few other combination operators:-

• endWith — When you want to ensure that a specific value is emitted after the source observable completes.
• startWith — When you need to provide an initial value to an observable sequence, ensuring that the consumer always receives a value upon subscription.
• zip — To combine values from multiple observables in a pairwise fashion

Creation

• interval — Emit numbers in sequence based on timeframe given

// RxJS v6+
import { interval } from 'rxjs';

//emit value in sequence every 1 second
const source = interval(1000);
//output: 0,1,2,3,4,5....
const subscribe = source.subscribe(val => console.log(val));

• timer — After given duration, emit numbers in sequence every specified duration until subscription is complete.

// RxJS v6+
import { timer } from 'rxjs';

/*
  timer takes a second argument, how often to emit subsequent values
  in this case we will emit first value after 1 second and subsequent
  values every 2 seconds after
*/
const source = timer(1000, 2000);
//output: 0,1,2,3,4,5......
const subscribe = source.subscribe(val => console.log(val));

• from — Converts argument(array, promise or iterator) to observable. If argument is array, all values will be emitted in sequence. Takes only one type of argument for the value unlike of.

//emit array as a sequence of values
const arraySource = from([1, 2, 3, 4, 5]);
//output: 1,2,3,4,5
const subscribe = arraySource.subscribe(val => console.log(val));

• of — Emit variable amount of values in a sequence and then emits a complete notification. This wraps value with observable and returns it.

// emits values of any type
const source = of({ name: 'Brian' }, [1, 2, 3], function hello() {
  return 'Hello';
});
// output: {name: 'Brian'}, [1,2,3], function hello() { return 'Hello' }
const subscribe = source.subscribe(val => console.log(val));
-----------------------------------------------------------------------

const source = of([1, 2, 3], [4, 5, 6]);
// output: [1,2,3], [4, 5, 6]
const subscribe = source.subscribe(val => console.log(val));

Filtering

• filter — Emit all values that pass certain condition.

// RxJS v6+
import { from } from 'rxjs';
import { filter } from 'rxjs/operators';

//emit (1,2,3,4,5)
const source = from([1, 2, 3, 4, 5]);
//filter out non-even numbers
const example = source.pipe(filter(num => num % 2 === 0));
//output: "Even number: 2", "Even number: 4"
const subscribe = example.subscribe(val => console.log(`Even number: ${val}`));

• find — Emit first item that passes given condition
• takeUntil — Emit values from source observable until notifier emits value

Scenario:- In below case, take values from observable until destroy is called.

destroy$ = new Subject<boolean>();

observable.pipe(takeUntil(this.destroy$))
            .subscribe();

ngOnDestroy() {
    this.destroy$.next();
}

• take — Take first n number of emissions given from observable and complete.

Scenario:- When we need only first value when user clicked on button or snapshot of value in a stream.

// RxJS v6+
import { of } from 'rxjs';
import { take } from 'rxjs/operators';

//emit 1,2,3,4,5
const source = of(1, 2, 3, 4, 5);
//take the first emitted value then complete
const example = source.pipe(take(1));
//output: 1
const subscribe = example.subscribe(val => console.log(val));

• skip — Skip given n number of emissions and take the next one. Opposite of take.

// RxJS v6+
import { interval } from 'rxjs';
import { skip } from 'rxjs/operators';

//emit every 1s
const source = interval(1000);
//skip the first 5 emitted values
const example = source.pipe(skip(5));
//output: 5...6...7...8........
const subscribe = example.subscribe(val => console.log(val));

• debounceTime- Discard emitted values that come in less than specified time between output. When you’re dealing with rapid sequences of events and only care about acting upon the last event after a specified duration, this can be used.

Scenario:- In searching, imagine you’re typing into a search box. Instead of firing off an API call with every keystroke , you’d want to wait a bit after the user stops typing to ensure you’re fetching data based on last emitted value after certain time.

Combining debounce with distinctUntilChanged will even improve it more to not send same request again and again if user is waiting for next keystroke even after debounce time.

this.myFormControl.valueChanges.pipe(
  debounceTime(300)
).subscribe(value => {
  // handle the value after 300ms of inactivity
});

Few Other filtering operators:-

• distinct- Emit distinct values based on previously emitted values.
• distinctUntilChanged- Only emit if current value is different from last emitted value.
• first/last — Emit the first or last value of sequence.
• single — Emit single value that meets the condition
• skip — Skip first n emissions. Opposite to take.
• skipUntil /skipWhile— Skip emissions until certain condition is met.
• takeLast — Take last n emissions.

Transformation

• map- Transforms values emitted from the source observable based on a provided projection function.

const numbers = [1, 2, 3, 4, 5];
const number$ = from(numbers); // we can use of as well
const numbersMultipliedByTen$ = number$.pipe(map(number => number * 10));
// 10, 20....50
numbersMultipliedByTen$.subscribe(console.log);

• concatMap — Used when you have an outer/source observable that emits values and, for each of those values, you want to execute another inner observable sequence, ensuring they are processed in order and not concurrently.

Scenario:- If you have a stream of user click events and for each click you want to initiate an HTTP request, but you need those requests to happen sequentially.

Get Naveen’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

For scenarios where you’d prefer to handle the most recent value and discard previous ones use switchMap.

For scenarios where you’re not concerned about the order of processing and just want everything to execute as it arrives use mergeMap.

//emit delay value
const source = of(2000, 1000);
// map value from source into inner observable, when complete emit result and move to next
const example = source.pipe(
  concatMap(val => of(`Delayed by: ${val}ms`).pipe(delay(val)))
);
//output: With concatMap: Delayed by: 2000ms, With concatMap: Delayed by: 1000ms
const subscribe = example.subscribe(val =>
  console.log(`With concatMap: ${val}`)
);

• mergeMap/flatMap- For each of emitted value from outer observable, it will create a new observable and flatten it with inner observable values.

let outerObservable= of(1,2);
let innerObservable= of('A','B');
 
outerObservable.pipe(
  mergeMap( val => {
    console.log('Outer value '+val)
    return innerObservable
  })
)
.subscribe(value => {
  console.log('Emitted ' + value);
})
 
 
//Output
Outer value 1
Emitted A
Emitted B
Outer value 2
Emitted A
Emitted B

• reduce- Applies an accumulator function over the source Observable, and returns the accumulated result when the source completes.

const source = of(1, 2, 3, 4);
const example = source.pipe(reduce((acc, val) => acc + val));
//output: Sum: 10'
const subscribe = example.subscribe(val => console.log('Sum:', val));

• switchMap — On each emission the previous inner observable is cancelled and the new observable is subscribed. You can remember this by the phrase switch to a new observable.

fromEvent(document, 'click')
  .pipe(
    // restart counter on every click
    switchMap(() => interval(1000))
    // if we dont use switchMap here, there will be many timer observables for every click getting added.
  )
  .subscribe(console.log);

• toArray — Collects all source emissions and emits them as an array when the source completes.

import { interval, take, toArray } from 'rxjs';

const source = interval(1000);
const example = source.pipe(
  take(10),
  toArray()
);

example.subscribe(value => console.log(value));

// output: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Utility

• tap — tap onto response of observable without subscribing to it. Operator is best for side effects: Useful in scenarios where you want to set response to service variable in service itself and use it directly in component without setting anything from next block in subscribe or for logging or throw/catch error based on response and let actual subscription flow as needed.

const source = of(1, 2, 3, 4, 5);
// transparently log values from source with 'tap'
const example = source.pipe(
  tap(val => console.log(`BEFORE MAP: ${val}`)),
  map(val => val + 10),
  tap(val => console.log(`AFTER MAP: ${val}`))
);

//'tap' does not transform values. it just logs above values
//output: 11...12...13...14...15
const subscribe = example.subscribe(val => console.log(val));

• delay — Allows you to hold back values for a specified duration before they’re released to subscribers.

const clicks = fromEvent(document, 'click');
const delayedClicks = clicks.pipe(delay(1000)); // each click emitted after 1 second
delayedClicks.subscribe(x => console.log(x));

• timeout — Error if Observable does not emit a value in given time span.

// simulate request
function makeRequest(timeToDelay) {
  return of('Request Complete!').pipe(delay(timeToDelay));
}

of(4000, 3000, 2000)
  .pipe(
    concatMap(duration =>
      makeRequest(duration).pipe(
        timeout(2500),
        catchError(error => of(`Request timed out after: ${duration}`))
      )
    )
  )
  /*
   *  "Request timed out after: 4000"
   *  "Request timed out after: 3000"
   *  "Request Complete!"
   */
  .subscribe(val => console.log(val));

• repeat — Repeat will output values from a source until the source completes, then it will resubscribe to the source a specified number of times, with a specified delay.

import { interval, take, repeat } from 'rxjs';

const source = interval(1000);
const result = source.pipe(take(3), repeat(2));

result.subscribe(x => console.log(x));

// Results every second
// 0
// 1
// 2
// 0
// 1
// 2

• timeInterval — Convert an Observable that emits items into one that emits indications of the amount of time elapsed between those emissions

Subjects

BehaviorSubjectis useful when you want to maintain and provide a “current value” to subscribers. when a new observer subscribes to a BehaviorSubject, it immediately receives the current value.

It requires an initial value upon instantiation. This is where it differs from a regular subject which doesn't have an initial value.

Contrasting with ReplaySubject while BehaviorSubject & ReplaySubject provide historical values, ReplaySubject can relay multiple previous values, not just the last one.

BehaviorSubject always requires an initial value and only stores the most recent value. Subscribers will get that latest value upon subscription. It's like walking into a movie and only catching the most recent scene.

subject, on the other hand, doesn't give new subscribers any previously emitted values. It’s the traditional live broadcast. If you're late, you miss out.

ReplaySubject can remember more than just the last scene. Depending on how it's configured, it might replay the entire movie or just the last few scenes.

// RxJS v6+
import { BehaviorSubject } from 'rxjs';

const subject = new BehaviorSubject(123);

// two new subscribers will get initial value => output: 123, 123
subject.subscribe(console.log);
subject.subscribe(console.log);

// two subscribers will get new value => output: 456, 456
subject.next(456);

// new subscriber will get latest value (456) => output: 456
subject.subscribe(console.log);

// all three subscribers will get new value => output: 789, 789, 789
subject.next(789);

// output: 123, 123, 456, 456, 456, 789, 789, 789

// RxJS v6+
import { Subject } from 'rxjs';

const sub = new Subject();

sub.next(1); // no subcriber yet to receive this value
sub.subscribe(x => {
  console.log('Subscriber A', x);
});
sub.next(2); // OUTPUT => Subscriber A 2
sub.subscribe(x => {
  console.log('Subscriber B', x);
});
sub.next(3); // OUTPUT => Subscriber A 3, Subscriber B 3 (logged from both subscribers)

// RxJS v6+
import { ReplaySubject } from 'rxjs';

const sub = new ReplaySubject(3);

sub.next(1);
sub.next(2);
sub.subscribe(console.log); // OUTPUT => 1,2
sub.next(3); // OUTPUT => 3
sub.next(4); // OUTPUT => 4
sub.subscribe(console.log); // OUTPUT => 2,3,4 (log of last 3 values from new subscriber)
sub.next(5); // OUTPUT => 5,5 (log from both subscribers)

Other operators

• share — If you want an observable which sends its data to the subscribers it has at the moment the data arrives, this is coined a hot observable, and one way to get a hot observable is to use the share operator.
• shareReplay — Useful in situations where you know you will have late subscribers to a stream that need access to previously emitted values. This ability to replay values on subscription is what differentiates share and shareReplay
• retry — Retry an observable sequence a specific number of times should an error occur. Useful for retrying HTTP requests.

If you only want to retry in certain cases, check out retryWhen

For non error cases where you want to repeat an observable, check out repeat.

// RxJS v6+
import { interval, of, throwError } from 'rxjs';
import { mergeMap, retry } from 'rxjs/operators';

//emit value every 1s
const source = interval(1000);
const example = source.pipe(
  mergeMap(val => { // ***mergeMap to merge and then do transform
    //throw error for demonstration
    if (val > 5) {
      return throwError('Error!');
    }
    return of(val);
  }),
  //retry 2 times on error
  retry(2)
);
/*
  output:
  0..1..2..3..4..5..
  0..1..2..3..4..5..
  0..1..2..3..4..5..
  "Error!: Retried 2 times then quit!"
*/
const subscribe = example.subscribe({
  next: val => console.log(val),
  error: val => console.log(`${val}: Retried 2 times then quit!`)
});

• catchError/throwError —

// RxJS v6+
import { throwError, of } from 'rxjs';
import { catchError } from 'rxjs/operators';
//emit error
const source = throwError('This is an error!');
//gracefully handle error, returning observable with error message
const example = source.pipe(catchError(val => of(`I caught: ${val}`)));
//output: 'I caught: This is an error'
const subscribe = example.subscribe(val => console.log(val));

References:

RxJS

Introduction | Learn RxJS

Yuvaraj S - Medium

My top 6 ‘Similar, but different’ RxJS operators

Complete Angular Tutorial For Beginners

Note:- While I try to maintain this as a part of good learning series, there is always room for improvement. I’m happy to make changes as needed for readers understanding and making it better.