Observables, Subjects, and BehaviorSubject

Lesson content

What is RxJS? RxJS is a reactive programming library used by Angular to work with asynchronous data, events, and value streams. Angular commonly uses RxJS for: HTTP requests Router events Form value changes User events WebSocket messages Shared application state Simple Definition RxJS represents asynchronous values as streams that can be observed, transformed, combined, and cancelled. Basic Flow Data Source ↓ Observable ↓ Operators ↓ Subscriber ↓ Value / Error / Complete What is an Observable? An Observable represents a stream of zero, one, or many values that may arrive immediately or over time. A consumer receives Observable values by subscribing to it. Basic Observable Example import { Observable } from 'rxjs'; const message$ = new Observable<string>( subscriber => { subscriber.next('Hello'); subscriber.next('Angular'); subscriber.complete(); } ); message$.subscribe({ next: value => console.log(value), error: error => console.error(error), complete: () => console.log('Completed') }); Output Hello Angular Completed Observable Notifications An Observable can send three types of notifications. Notification Purpose next() Sends a value error() Sends an error and ends the stream complete() Ends the stream successfully Important Rule After an Observable errors or completes, it cannot emit more values. Observable Naming Convention Observable variable names commonly end with the $ symbol. users$ products$ routeParams$ searchResults$ The dollar sign is only a naming convention. It is not required by JavaScript or Angular. Cold Observable Most Observables are cold by default. A cold Observable starts its producer separately for every subscriber. import { Observable } from 'rxjs'; const random$ = new Observable<number>( subscriber => { subscriber.next( Math.random() ); } ); random$.subscribe( value => console.log('A:', value) ); random$.subscribe( value => console.log('B:', value) ); Each subscriber can receive a different random value because the producer runs separately. Cold Observable Flow Subscriber A ↓ New Producer ↓ Value A Subscriber B ↓ New Producer ↓ Value B Observable with Timer import { interval } from 'rxjs'; const counter$ = interval(1000); const subscription = counter$.subscribe( value => console.log(value) ); The Observable emits a new number every second until the subscription is cancelled. Subscription A subscription represents the execution of an Observable. Use unsubscribe() to stop receiving values and release associated resources. const subscription = counter$.subscribe( value => console.log(value) ); subscription.unsubscribe(); Observable Teardown An Observable can return a cleanup function that runs when the subscription is cancelled. const timer$ = new Observable<number>( subscriber => { let count = 0; const timerId = window.setInterval( () => { subscriber.next(count++); }, 1000 ); return () => { window.clearInterval( timerId ); }; } ); Creating Observables RxJS provides several functions for creating Observables. Function Purpose of() Emits provided values from() Creates a stream from an array, Promise, or iterable interval() Emits increasing numbers repeatedly timer() Emits after a delay fromEvent() Creates a stream from DOM or event-source events of() Example import { of } from 'rxjs'; const numbers$ = of(10, 20, 30); numbers$.subscribe( value => console.log(value) ); from() Example import { from } from 'rxjs'; const users$ = from([ 'Ajay', 'Sneha', 'Rahul' ]); users$.subscribe( user => console.log(user) ); fromEvent() Example import { fromEvent } from 'rxjs'; const button = document.querySelector( '#saveButton' ); if (button) { const click$ = fromEvent( button, 'click' ); click$.subscribe( () => console.log('Clicked') ); } Observable Operators Operators transform, filter, combine, or control Observable streams. Source Observable ↓ Operator 1 ↓ Operator 2 ↓ Subscriber pipe() The pipe() method applies operators to an Observable. import { filter, map, of } from 'rxjs'; const numbers$ = of(1, 2, 3, 4, 5); numbers$ .pipe( filter( value => value % 2 === 0 ), map( value => value * 10 ) ) .subscribe( value => console.log(value) ); Output 20 40 Common RxJS Operators Operator Purpose map() Transforms every emitted value filter() Allows only matching values tap() Runs side effects without changing values debounceTime() Waits for input activity to pause distinctUntilChanged() Skips consecutive duplicate values switchMap() Switches to a new inner Observable and cancels the previous one catchError() Handles stream errors take() Takes a limited number of values takeUntil() Completes when another Observable emits shareReplay() Shares and replays previous emissions Observable in Angular HTTP Angular's HttpClient methods return Observables. import { HttpClient } from '@angular/common/http'; import { inject, Injectable } from '@angular/core'; import { Observable } from 'rxjs'; export interface User { id: number; name: string; } @Injectable({ providedIn: 'root' }) export class UserService { private http = inject(HttpClient); getUsers(): Observable<User[]> { return this.http.get<User[]>( '/api/users' ); } } Using the HTTP Observable export class UsersComponent { private userService = inject(UserService); users$ = this.userService.getUsers(); } Template with Async Pipe @if (users$ | async; as users) { @for (user of users; track user.id) { <p>{{ user.name }}</p> } } Async Pipe The async pipe subscribes to an Observable, returns its latest value, updates the template, and automatically unsubscribes when the view is destroyed. <p> {{ username$ | async }} </p> Prefer the async pipe when an Observable is used only by the template. What is a Subject? A Subject is both an Observable and an Observer. It can: Receive values using next() Emit values to multiple subscribers Send errors Complete the stream Subject Flow Producer ↓ Subject.next(value) ↓ ┌───────────────┐ ↓ ↓ Subscriber A Subscriber B Subject Example import { Subject } from 'rxjs'; const notification$ = new Subject<string>(); notification$.subscribe( message => console.log( 'Subscriber A:', message ) ); notification$.subscribe( message => console.log( 'Subscriber B:', message ) ); notification$.next( 'User logged in' ); Output Subscriber A: User logged in Subscriber B: User logged in Subject is Multicast A Subject multicasts the same value to all active subscribers. One Subject ↓ Same Emission ↓ Multiple Subscribers Subject Does Not Store a Previous Value A normal Subject sends values only to subscribers that are currently subscribed. const subject = new Subject<number>(); subject.next(10); subject.subscribe( value => console.log(value) ); subject.next(20); Output 20 The subscriber does not receive 10 because it subscribed after that value was emitted. Subject in an Angular Service import { Injectable } from '@angular/core'; import { Subject } from 'rxjs'; @Injectable({ providedIn: 'root' }) export class NotificationService { private messageSubject = new Subject<string>(); readonly message$ = this.messageSubject .asObservable(); sendMessage( message: string ) { this.messageSubject.next( message ); } } Send a Message export class SenderComponent { private notificationService = inject(NotificationService); send() { this.notificationService .sendMessage( 'Order created' ); } } Receive a Message export class HeaderComponent { private notificationService = inject(NotificationService); message$ = this.notificationService .message$; } Why Use asObservable()? Use asObservable() to expose a Subject as a read-only Observable. private userSubject = new Subject<User>(); readonly user$ = this.userSubject .asObservable(); External consumers can subscribe, but they cannot directly call next() . What is BehaviorSubject? A BehaviorSubject is a special type of Subject that stores the latest value. It requires an initial value and immediately emits the current value to every new subscriber. BehaviorSubject Example import { BehaviorSubject } from 'rxjs'; const userStatus$ = new BehaviorSubject<string>( 'Offline' ); userStatus$.subscribe( value => console.log( 'Subscriber A:', value ) ); userStatus$.next( 'Online' ); userStatus$.subscribe( value => console.log( 'Subscriber B:', value ) ); Output Subscriber A: Offline Subscriber A: Online Subscriber B: Online Subscriber B immediately receives Online because it is the latest stored value. BehaviorSubject Flow Initial Value ↓ BehaviorSubject ↓ Subscriber A receives current value ↓ New value emitted ↓ Latest value stored ↓ Subscriber B joins ↓ Subscriber B receives latest value Read the Current Value BehaviorSubject exposes its latest value through value or getValue() . const status$ = new BehaviorSubject( 'Offline' ); console.log( status$.value ); console.log( status$.getValue() ); In reactive application code, prefer subscribing or exposing the stream rather than repeatedly reading the value imperatively. Subject vs BehaviorSubject Subject BehaviorSubject Does not require an initial value Requires an initial value Does not store the latest value Stores the latest value New subscribers receive only future values New subscribers immediately receive the latest value Useful for events Useful for current state No direct current-value access Provides value and getValue() Observable vs Subject vs BehaviorSubject Feature Observable Subject BehaviorSubject Can be subscribed to Yes Yes Yes Can manually emit with next() No Yes Yes Multicast by default No Yes Yes Requires initial value No No Yes Stores latest value No No Yes Best used for Async data stream Events and notifications Current shared state BehaviorSubject for Shared State import { Injectable } from '@angular/core'; import { BehaviorSubject } from 'rxjs'; export interface User { id: number; name: string; } @Injectable({ providedIn: 'root' }) export class UserStateService { private userSubject = new BehaviorSubject<User | null>( null ); readonly user$ = this.userSubject .asObservable(); setUser( user: User ) { this.userSubject.next( user ); } clearUser() { this.userSubject.next( null ); } } Use Shared State export class ProfileComponent { private userState = inject(UserStateService); user$ = this.userState.user$; } Template @if (user$ | async; as user) { <p>{{ user.name }}</p> } @else { <p>No user logged in</p> } ReplaySubject ReplaySubject stores and replays a configurable number of previous values to new subscribers. import { ReplaySubject } from 'rxjs'; const messages$ = new ReplaySubject<string>(2); messages$.next('First'); messages$.next('Second'); messages$.next('Third'); messages$.subscribe( value => console.log(value) ); Output Second Third AsyncSubject AsyncSubject emits only the final value when the Subject completes. import { AsyncSubject } from 'rxjs'; const result$ = new AsyncSubject<number>(); result$.subscribe( value => console.log(value) ); result$.next(10); result$.next(20); result$.next(30); result$.complete(); Output 30 Subject Types Comparison Type Behaviour Subject Emits only future values to current subscribers BehaviorSubject Emits the current latest value to new subscribers ReplaySubject Replays a configured number of previous values AsyncSubject Emits the last value only after completion Unsubscribing in Angular Long-running subscriptions should be cleaned up when a component is destroyed. Using takeUntilDestroyed() import { Component, DestroyRef, inject } from '@angular/core'; import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; @Component({ selector: 'app-users', standalone: true, template: ` <p>Users</p> ` }) export class UsersComponent { private destroyRef = inject(DestroyRef); private userService = inject(UserService); constructor() { this.userService .getUsers() .pipe( takeUntilDestroyed( this.destroyRef ) ) .subscribe( users => console.log(users) ); } } Using takeUntilDestroyed() in an Injection Context constructor() { this.userService .getUsers() .pipe( takeUntilDestroyed() ) .subscribe(); } When called inside an Angular injection context, the DestroyRef argument can be omitted. Subscriptions That Often Do Not Need Manual Cleanup Subscriptions managed by the async pipe Subscriptions created through toSignal() with default cleanup Finite HTTP request Observables that complete after receiving the response Framework-managed router and forms streams when used through managed mechanisms Even when an Observable completes automatically, avoid unnecessary nested subscriptions and duplicated requests. Search Example with Observable import { Component, inject } from '@angular/core'; import { FormControl, ReactiveFormsModule } from '@angular/forms'; import { debounceTime, distinctUntilChanged, switchMap } from 'rxjs'; @Component({ selector: 'app-search', standalone: true, imports: [ ReactiveFormsModule ], template: ` <input [formControl]="searchControl" placeholder="Search" /> ` }) export class SearchComponent { private productService = inject(ProductService); searchControl = new FormControl( '', { nonNullable: true } ); results$ = this.searchControl .valueChanges .pipe( debounceTime(300), distinctUntilChanged(), switchMap( searchTerm => this.productService .searchProducts( searchTerm ) ) ); } Search Flow User Types ↓ valueChanges Observable ↓ debounceTime() ↓ distinctUntilChanged() ↓ switchMap() ↓ Cancel Previous Request ↓ Start Latest Search ↓ Display Results Avoid Nested Subscriptions Avoid this.route.paramMap.subscribe( params => { const id = params.get('id'); this.userService .getUser(id!) .subscribe( user => { this.user = user; } ); } ); Prefer user$ = this.route.paramMap.pipe( map( params => params.get('id')! ), switchMap( id => this.userService .getUser(id) ) ); BehaviorSubject vs Signal BehaviorSubject Signal RxJS stream Angular reactive state primitive Read through subscription Read synchronously by calling it Updated with next() Updated with set() or update() Useful for stream composition Useful for component and shared UI state Supports all RxJS operators Uses computed() and effect() Emits notifications over time Always exposes a current value Use Signals for straightforward reactive UI state. Use RxJS for asynchronous streams, events, cancellation, WebSockets, and complex operator-based workflows. Convert Observable to Signal Use toSignal() to expose the latest Observable value as a Signal. import { toSignal } from '@angular/core/rxjs-interop'; users = toSignal( this.userService.getUsers(), { initialValue: [] } ); By default, Angular automatically cleans up the subscription when the creating component or service is destroyed. BehaviorSubject with requireSync private statusSubject = new BehaviorSubject( 'Offline' ); status = toSignal( this.statusSubject, { requireSync: true } ); Because BehaviorSubject emits immediately when subscribed, requireSync: true can guarantee that the Signal always has a value. Convert Signal to Observable import { signal } from '@angular/core'; import { toObservable } from '@angular/core/rxjs-interop'; searchTerm = signal(''); searchTerm$ = toObservable( this.searchTerm ); Use RxJS Operators results$ = this.searchTerm$ .pipe( debounceTime(300), distinctUntilChanged(), switchMap( term => this.productService .searchProducts(term) ) ); Common Mistakes 1. Making Subjects Public Avoid: userSubject = new BehaviorSubject<User | null>( null ); Prefer: private userSubject = new BehaviorSubject<User | null>( null ); readonly user$ = this.userSubject .asObservable(); 2. Forgetting to Unsubscribe Avoid unmanaged long-running subscriptions. Prefer: stream$ .pipe( takeUntilDestroyed() ) .subscribe(); 3. Using BehaviorSubject for One-Time Events A BehaviorSubject replays its latest value to new subscribers, which may repeat an old notification. Use a normal Subject for one-time events such as: Toast messages Button events Refresh commands Logout notifications 4. Nested Subscriptions Use higher-order operators such as switchMap() , mergeMap() , concatMap() , or exhaustMap() instead of subscribing inside another subscription. 5. Calling subscribe() Repeatedly Multiple subscriptions to a cold Observable may repeat API requests or producer work. Share the stream when multiple consumers genuinely need the same execution. Best Practices Use Observables for asynchronous data streams Use a Subject for events and notifications Use a BehaviorSubject when subscribers require the latest current value Keep Subjects private and expose read-only Observables Prefer the async pipe for template subscriptions Use takeUntilDestroyed() for manual long-running subscriptions Use RxJS operators instead of nested subscriptions Handle errors with catchError() Use switchMap() when previous async work should be cancelled Use Signals for simple reactive UI state when stream operators are unnecessary Interview Questions What is an Observable? An Observable represents a stream of values that can arrive synchronously or asynchronously over time. What is a Subject? A Subject is both an Observable and an Observer. It can manually receive values using next() and multicast them to multiple subscribers. What is a BehaviorSubject? A BehaviorSubject is a Subject that requires an initial value, stores the latest value, and immediately emits that value to new subscribers. What is the difference between a cold Observable and a Subject? A cold Observable normally starts a separate producer for every subscriber. A Subject shares the same emissions with all current subscribers. When should you use a Subject? Use a Subject for events, notifications, commands, or values that should be sent only to currently active subscribers. When should you use a BehaviorSubject? Use a BehaviorSubject when the application needs a current value and new subscribers must immediately receive the latest state. Does an HTTP Observable require manual unsubscribe? A normal Angular HTTP Observable completes after the response, so manual cleanup is usually unnecessary. Long-running streams still require managed cleanup. Why should Subjects be private? Keeping the Subject private prevents external classes from calling next() and changing service state unexpectedly. Do Signals replace Observables? No. Signals are useful for current reactive state, while Observables are well suited for asynchronous streams, cancellation, events, WebSockets, and complex stream composition. Key Point Observable represents a value stream, Subject broadcasts events, and BehaviorSubject broadcasts while also remembering the latest value. Easy Formula Observable = Async Value Stream Subject = Observable + Manual Emission + Multicast BehaviorSubject = Subject + Initial Value + Latest Value ReplaySubject = Subject + Previous Values AsyncSubject = Final Value After Completion Producer ↓ Observable or Subject ↓ Operators ↓ Subscribers ↓ Next / Error / Complete