1. Key Observable Creation Functions

1.1 of operator

of operator is used to create an observablt to emit a sequence of values.

const temp$ = of(1,2,3).subscribe(console.log);

1.2 from

Converts an iteratable into observable

from([4,5,6]).subscribe(console.log)        // emits 4, 5, 6

1.3 interval

Sends a number after every interval

const source$ = interval(1000);

const subscription= source$.subscribe((x)=>{
console.log(x);     // log 1, 2, 3, 4 ...... after every 1000 millisecond
});

1.4 timer

timer(dueTime, period?): Emits one value after dueTime, then optionally continues to emit values every period milliseconds

timer(2000).subscribe(val => console.log(val)); // 0 after 2 seconds
timer(1000, 500).subscribe(val => console.log(val)); // 0 after 1s, then 1 after 0.5s, etc.

2. Commonly Used Operators (Piping)

2.1 Transformation Operators

2.1.1 map

Use to transform the value received from observable

from([1,2,3]).pipe(
    map(x => x*10 )
)
.subscribe(x => {
    console.log(x);     // logs 10,20 30
})

2.1.2 pluck

Use to extract a property from an observable, but deprecated in favor of map

of({ name: 'Alice', age: 30 })
.pipe(pluck('name'))
.subscribe(x=>{
    console.log (x);        // Alice
})

2.1.3 scan

scan(accumulator, seed?): Applies an accumulator function over the source Observable, and returns each intermediate result. Like reduce, but emits every step.

of(1, 2, 3, 4).pipe(
  scan((acc, value) => acc + value, 0)
).subscribe(console.log);

• 0 + 1 = 1 → emitted
• 1 + 2 = 3 → emitted
• 3 + 3 = 6 → emitted
• 6 + 4 = 10 → emitted

2.2 Filtering Operators

2.2.1 filter

filter(predicate): Emits only those values from the source Observable that satisfy a specified predicate function

of(1, 2, 3, 4).pipe(
  filter(x => x%2 === 0)
).subscribe(console.log);   // 2,4

2.2.2 take(count)

Emits only the first count values emitted by the source Observable.

of(1, 2, 3, 4).pipe(
  take(2)
).subscribe(console.log);   // 1,2

2.2.3 takeUntil(notifier)

takeUntil(notifier): Emits values until the notifier Observable emits a value

 interval(1000).pipe(
      takeUntil(timer(5000)) // Emits for 5 seconds
    ).subscribe(val => console.log(val));       // 0, 1, 2, 3

Another example

private destroy$ = new Subject<void>();

ngOnInit() {
  observable$.pipe(
    takeUntil(this.destroy$)
  ).subscribe(data => console.log(data));
}

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

2.2.4 debounceTime

2.2.5 distinctUntilChanged

Emits all values that are distinct by comparison from the previous value.

of(1, 1, 2, 2, 1, 3).pipe(
  distinctUntilChanged()
).subscribe(val => console.log(val)); // 1, 2, 1, 3

2.2.6 first

first(predicate?): Emits only the first value (or the first value that satisfies a predicate). Completes immediately.

of(10,20,30).pipe(
    first(x=> x === 20)
)
.subscribe(x=> console.log(x)); // 20

2.2.7 last

last(predicate?): Emits only the last value (or the last value that satisfies a predicate) before the observable completes.

of(10,20,30).pipe(
    last()
)
.subscribe(x=> console.log(x)); // 30

2.2.8 skip

Skip a certain number of emitted values before continuing to process them. Useful for ignoring default values or headers.

of(1, 2, 3, 4, 5).pipe(
    skip(2)
)
.subscribe(x => {
    console.log(x);     // 3,4,5
});

2.2.9 takeWhile

The takeWhile operator in RxJS is used to emit values from an observable until a condition fails—after which, the observable completes immediately.

takeWhile(predicate: (value, index) => boolean)

import { of } from 'rxjs';
import { takeWhile } from 'rxjs/operators';

of(1, 2, 3, 4, 5, 6, 7).pipe(
  takeWhile(value => value < 5)
).subscribe(console.log); // 1, 2, 3, 4

As soon as the value 5 arrives, the condition value < 5 fails, so the stream completes and 5 is not emitted.

2.3 Combination Operators

2.3.1 concat

concat(...observables):

• Subscribes to Observables one after the other, in sequence.
• Waits for one to complete before subscribing to the next.

concat(of(1,2), of(3,4)).subscribe(x=> console.log(x));     // 1, 2, 3, 4

2.3.2 zip (Pair items by index (like a zipper))

Imagine you have two zippers with teeth — one red and one blue. zip joins first red with first blue, second red with second blue, and so on.

• 📌 zip emits values only when all input observables emit at the same index.
• Stops at the shortest observable.

import { zip, of } from 'rxjs';

const names$ = of('Alice', 'Bob', 'Charlie');
const ages$ = of(25, 30, 35);

zip(names$, ages$).subscribe(([name, age]) => {
  console.log(`${name} is ${age} years old`);
});

2.3.3 merge Mix items as they arrive (like merging two rivers)

merge(...observables): Subscribes to all inner Observables concurrently and merges their emissions into a single Observable. Imagine two rivers flowing into one stream — whenever water comes from either side, it flows together into the same stream.

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

const fast$ = interval(500).pipe(map(x => `Fast: ${x}`), take(3));
const slow$ = interval(1000).pipe(map(x => `Slow: ${x}`), take(2));

merge(fast$, slow$).subscribe(val => console.log(val));

• Fast: 0
• Slow: 0
• Fast: 1
• Fast: 2
• Slow: 1

2.3.4 combineLatest: Combine latest from all when any emits

You're watching two stock prices. When either changes, you want the latest price of both before making a decision.

combineLatest waits until all observables have emitted at least once, then emits whenever any observable emits, using the latest value of each.

const temperature$ = interval(1000).pipe(map(x => `Temp ${20 + x}`), take(3));
const humidity$ = interval(1500).pipe(map(x => `Humidity ${50 + x}`), take(2));

combineLatest([temperature$, humidity$]).subscribe(([temp, humidity]) => {
    console.log(`${temp}, ${humidity}`);
});

• Temp 20, Humidity 50 // time 1500
• Temp 21, Humidity 50 // time 2000
• Temp 22, Humidity 50 // time 3000
• Temp 22, Humidity 51 // time 3000

2.3.4 forkJoin

forkJoin waits for all observables to complete and then emits one array (or object) with the last emitted value from each observable

import { forkJoin } from 'rxjs';

forkJoin([
      getUser(),
      getNotifications(),
      getOrders(),
    ])
    .subscribe(([userResponse,notificationResponse,orderResponse]) => {

    });

2.4 Utility Operators

2.4.1 tap

• The tap operator is used to perform side effects for notifications from the source observable — without affecting the stream.
• To log, debug, or trigger side effects (like analytics) without modifying the actual data flowing through the observable.

import { of } from 'rxjs';
import { tap, map } from 'rxjs/operators';

of(1, 2, 3)
  .pipe(
    tap(value => console.log('Before map:', value)),
    map(x => x * 10),
    tap(value => console.log('After map:', value))
  )
  .subscribe();

• Before map: 1
• After map: 10
• Before map: 2
• After map: 20
• Before map: 3
• After map: 30

2.4.2 delay

delay(dueTime): Delays the emission of values from the source Observable by a specified dueTime.

import { of } from 'rxjs';
import { delay } from 'rxjs/operators';
of(1, 2).pipe(
  delay(1000)
).subscribe(val => console.log(val)); // Emits 1 after 1s, then 2 after 1s from when it was scheduled

2.5 Error Handling Operators

2.5.1 catchError(selector)

Catches errors on the source Observable and returns a new Observable, or throws an error

import { of, throwError } from 'rxjs';
import { map, catchError } from 'rxjs/operators';
of(1, 2, 3).pipe(
  map(val => {
    if (val === 2) throw 'Error at 2';
    return val;
  }),
  catchError(err => {
    console.error('Caught error:', err);
    return of(0); // Recover with a default value
  })
).subscribe(val => console.log(val)); // 1, Caught error: Error at 2, 0

3. Higher-Order Mapping Operators (Flattening Operators)

3.1 switchMap

3.2 concatMap

3.3 mergeMap

3.4 exhaustMap

4. Subjects

A Subject in RxJS is a special type of Observable that allows values to be multicasted to many Observers. In simpler terms, a Subject is both:

• an Observable (you can subscribe() to it)
• an Observer (you can send values to it with next())

Understand the difference between unicast and multicast

Analogy

• Unicast = Everyone cooks their own meal from scratch.
• Multicast = One meal is cooked and shared with everyone.

🔁 Observable restarts its producer logic for each subscriber → Unicast

import { Observable } from 'rxjs';

const obs$ = new Observable(subscriber => {
 console.log('Observable started');
 subscriber.next(Math.random()); // emits a new random number for each subscriber
});

obs$.subscribe(val => console.log('Subscriber A:', val));
obs$.subscribe(val => console.log('Subscriber B:', val));

/*
Observable started
Subscriber A: 0.5824...

Observable started
Subscriber B: 0.9328...
*/

Multicast Example – Using a Subject

🎯 Subject shares the same value with all subscribers → Multicast

import { Subject } from 'rxjs';

const subject$ = new Subject<number>();

// same random value is pushed once
subject$.subscribe(val => console.log('Subscriber A:', val));
subject$.subscribe(val => console.log('Subscriber B:', val));

subject$.next(Math.random());

/*
Subscriber A: 0.7352...
Subscriber B: 0.7352...
*/

Subject vs BehaviorSubject

• Subject has no memory, any subscribers will only get the value once it is emitted
• SubjectBehavior maintain the last value and require an initial value upon intialization, once a subscriber subscribe to it, it will get the last value automatically

Example of Subject

import { Subject } from 'rxjs';

const subject = new Subject<number>();

subject.subscribe(val => console.log('A:', val));

subject.next(1); // A: 1
subject.next(2); // A: 2

subject.subscribe(val => console.log('B:', val)); // 👈 Subscribes later

subject.next(3); // A: 3, B: 3

/*
A: 1
A: 2
A: 3
B: 3
*/

Example of BehaviorSubject

import { BehaviorSubject } from 'rxjs';

const behavior = new BehaviorSubject<number>(0); // 👈 Must have an initial value

behavior.subscribe(val => console.log('A:', val));

behavior.next(1); // A: 1
behavior.next(2); // A: 2

behavior.subscribe(val => console.log('B:', val)); // 👈 Subscribes later

behavior.next(3); // A: 3, B: 3

/*
A: 0
A: 1
A: 2
B: 2
A: 3
B: 3
*/

ReplaySubject vs AsyncBehavior

Example of ReplaySubject

import { ReplaySubject } from 'rxjs';

const replay$ = new ReplaySubject<number>(2); // Keep last 2 values

replay$.next(1);
replay$.next(2);
replay$.next(3);

replay$.subscribe(val => console.log('Subscriber A:', val));

replay$.next(4);

/*
Subscriber A: 2
Subscriber A: 3
Subscriber A: 4
*/

Example of AsyncSubject

Only emit the value once it is completed

import { AsyncSubject } from 'rxjs';

const async$ = new AsyncSubject<number>();

async$.next(1);
async$.next(2);
async$.next(3);

async$.subscribe(val => console.log('Subscriber A:', val));

async$.next(4);

async$.complete(); // 👈 Only now will it emit the last value

/*
Subscriber A: 4
*/

5. Hot vs Cold Observables

This distinction is about when and how the observable starts emitting values and who receives them.

Cold Observable

A cold observable starts emitting values only when subscribed to. Each subscriber gets its own independent execution

• Values are generated per subscriber.
• Ideal for unicast operations (like HTTP requests, setTimeout, interval, etc.).
• Think of a Netflix movie: everyone gets their own stream from the beginning.

const cold$ = new Observable(observer => {
  console.log("Observable started");
  observer.next(Math.random());
});

cold$.subscribe(val => console.log("Subscriber 1:", val));
cold$.subscribe(val => console.log("Subscriber 2:", val));

/*
Observable started
Subscriber 1: 0.84
Observable started
Subscriber 2: 0.51
*/

Hot Observable

hot observable starts emitting values regardless of subscriptions. Subscribers share the same execution and get data from that point onward.

Characteristics:

• Values are shared across subscribers.
• Ideal for multicast sources (like user clicks, WebSocket, live timer).
• Think of a live YouTube stream: join late, you miss the beginning.

const subject = new Subject();

subject.subscribe(val => console.log("Subscriber 1:", val));
subject.next(Math.random());

subject.subscribe(val => console.log("Subscriber 2:", val));
subject.next(Math.random());


/*
Subscriber 1: 0.72
Subscriber 1: 0.33
Subscriber 2: 0.33
*/

Here both subscribers got the same value on the second .next().