RxJS for Angular Developers: Observables, Operators & Real-World Patterns

import { of, from, interval, fromEvent, Observable, timer } from 'rxjs';
// of() - emits each argument as a separate value, then completes
const status$ = of('loading', 'success', 'complete');
status$.subscribe(s => console.log(s));
// Logs: 'loading', 'success', 'complete'
// of() is commonly used in tests and for fallback values
const mockProducts$ = of([{ id: 1, name: 'Widget' }]);
// from() - converts Promises, arrays, or iterables to Observables
const fromArray$ = from([1, 2, 3, 4, 5]); // emits each element
const fromPromise$ = from(fetch('https://api.example.com/data'));
// interval() - emits incrementing numbers at a fixed interval (ms)
const clock$ = interval(1000); // emits 0, 1, 2... every second
// Used for polling: check for updates every 30 seconds
// timer() - waits delay, then emits (optionally on an interval)
const delayed$ = timer(2000);          // emits once after 2 seconds
const polling$ = timer(0, 30000);      // starts immediately, repeats every 30s
// fromEvent() - wraps DOM events as an Observable
const clicks$ = fromEvent(document, 'click');
const keyups$ = fromEvent(document, 'keyup');
// new Observable() - manual creation for custom async operations
const custom$ = new Observable<number>(observer => {
observer.next(1);
observer.next(2);
observer.complete();
// Return cleanup function (called on unsubscribe)
return () => console.log('Cleaned up');
});

import { map, filter, tap, catchError, of } from 'rxjs';
// map() - transforms each emitted value
// Use: extract data, convert types, shape API responses
productService.getAll().pipe(
map(response => response.data),           // extract data array from paginated response
map(products => products.filter(p => p.inStock)), // filter in-stock only
map(products => products.map(this::toDto)) // convert entities to DTOs
);
// filter() - emits only values that pass the predicate
// Use: remove null/undefined, filter by condition
route.paramMap.pipe(
map(params => params.get('id')),
filter((id): id is string => id !== null) // type guard - narrow type
);
// tap() - side effect without modifying the stream
// Use: logging, updating loading state, triggering side effects
productService.getAll().pipe(
tap(() => this.isLoading = true),
tap({ error: () => this.isLoading = false }),
tap(() => this.isLoading = false)
);

import { switchMap, mergeMap, concatMap, exhaustMap } from 'rxjs';
// switchMap() - CANCELS the previous inner Observable when new value arrives
// Perfect for: search-as-you-type, route parameter changes, any "latest wins" scenario
searchInput.valueChanges.pipe(
debounceTime(300),
switchMap(term =>
productService.search(term) // if user types again before results arrive, CANCEL and restart
)
);
// Real Angular route param handling - switchMap is essential here
route.paramMap.pipe(
switchMap(params => {
const id = +params.get('id')!;
return productService.getProduct(id); // cancel previous request if route changes
})
);
// mergeMap() - runs ALL inner Observables concurrently (no cancellation)
// Perfect for: parallel operations where all results matter
const productIds = [1, 2, 3, 4, 5];
from(productIds).pipe(
mergeMap(id => productService.getProduct(id)) // all 5 requests fire simultaneously
);
// concatMap() - queues inner Observables, runs ONE at a time in ORDER
// Perfect for: sequential operations where order matters
from(itemsToSave).pipe(
concatMap(item => orderService.save(item)) // save items one by one, in order
);
// exhaustMap() - IGNORES new values while inner Observable is still running
// Perfect for: login button - ignore additional clicks while request is in flight
loginButton.clicks$.pipe(
exhaustMap(() => authService.login(credentials)) // ignore double-clicks
);

import { Subject, BehaviorSubject, ReplaySubject } from 'rxjs';
// Subject - NO initial value, late subscribers miss past emissions
const events$ = new Subject<string>();
events$.next('login');      // emitted BEFORE anyone subscribed
events$.subscribe(e => console.log(e)); // won't see 'login'
events$.next('purchase');  // subscriber will see 'purchase'
// BehaviorSubject - REQUIRES initial value, new subscribers get current value
const user$ = new BehaviorSubject<User | null>(null); // null = not logged in
user$.subscribe(u => console.log('A sees:', u));   // A immediately sees null
user$.next({ id: '1', name: 'Alice' });           // A sees Alice
user$.subscribe(u => console.log('B sees:', u));   // B immediately sees Alice
console.log(user$.getValue());  // synchronous read: { id: '1', name: 'Alice' }
// ReplaySubject - buffers N past values, new subscribers see them all
const log$ = new ReplaySubject<string>(3); // buffer last 3 values
log$.next('Error 1');
log$.next('Error 2');
log$.next('Error 3');
log$.next('Error 4');
log$.subscribe(e => console.log(e)); // sees Error 2, 3, 4 (last 3)
// BehaviorSubject state service pattern - the most common Angular pattern
@Injectable({ providedIn: 'root' })
export class UserStateService {
private readonly _user$ = new BehaviorSubject<User | null>(null);
// asObservable() hides next() - components can read but not push
readonly user$        = this._user$.asObservable();
readonly isLoggedIn$  = this.user$.pipe(map(u => u !== null));
readonly displayName$ = this.user$.pipe(map(u => u?.name ?? 'Guest'));
setUser(user: User)  { this._user$.next(user); }
clearUser()           { this._user$.next(null); }
getCurrentUser()      { return this._user$.getValue(); }
}

import { forkJoin, combineLatest, zip } from 'rxjs';
// forkJoin() - waits for ALL to complete, emits their LAST values as an array
// Perfect for: parallel API calls where you need ALL results before proceeding
@Injectable({ providedIn: 'root' })
export class DashboardService {
loadDashboard() {
// All 3 requests fire simultaneously - result arrives when all 3 complete
return forkJoin({
products:      productService.getProducts(),
orders:        orderService.getRecentOrders(),
notifications: notifService.getUnread()
}).pipe(
map(({ products, orders, notifications }) => ({
productCount:  products.total,
recentOrders:  orders.data,
unreadCount:   notifications.length
}))
);
}
}
// combineLatest() - emits when ANY source emits, with latest from ALL sources
// Perfect for: derived values that depend on multiple live streams
readonly filteredProducts$ = combineLatest([
this.allProducts$,   // BehaviorSubject - emits on product changes
this.activeFilter$,  // BehaviorSubject - emits on filter changes
this.searchTerm$     // BehaviorSubject - emits on search input
]).pipe(
map(([products, filter, term]) =>
products
.filter(p => filter === 'all' || p.category === filter)
.filter(p => p.name.toLowerCase().includes(term.toLowerCase()))
)
);
// filteredProducts$ automatically recalculates when ANY of the 3 sources emit
// combineLatest with route params and auth state - extremely common pattern
readonly pageData$ = combineLatest([
route.paramMap.pipe(map(p => p.get('id'))),
userStateService.user$
]).pipe(
switchMap(([productId, user]) =>
productService.getProductForUser(productId!, user?.id)
)
);

import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; // Angular 16+
import { Subject, takeUntil }  from 'rxjs';
// APPROACH 1: async pipe - ALWAYS use in templates
@Component({
template:     <!-- Angular subscribes and unsubscribes automatically -->     <div *ngFor="let p of products$ | async">{{ p.name }}</div>     <p>{{ userDisplayName$ | async }}</p>  
})
export class ProductsComponent {
products$        = productService.getAll();
userDisplayName$ = userService.displayName$;
// No ngOnDestroy needed - async pipe handles everything
}
// APPROACH 2: takeUntilDestroyed() - Angular 16+ (recommended for imperative)
@Component({ selector: 'app-notifications', standalone: true, template: ... })
export class NotificationsComponent implements OnInit {
private destroyRef = inject(DestroyRef);
notifications: Notification[] = [];
ngOnInit() {
notificationService.notifications$.pipe(
takeUntilDestroyed(this.destroyRef) // auto-unsubscribes on destroy
).subscribe(n => this.notifications = n);
}
}
// APPROACH 3: takeUntil(destroy$) - pre-Angular 16, still valid
@Component({ selector: 'app-clock', standalone: true, template: {{ time }} })
export class ClockComponent implements OnInit, OnDestroy {
private destroy$ = new Subject<void>();
time = '';
ngOnInit() {
interval(1000).pipe(
takeUntil(this.destroy$) // interval runs until component is destroyed
).subscribe(() => {
this.time = new Date().toLocaleTimeString();
});
}
ngOnDestroy() {
this.destroy$.next();
this.destroy$.complete();
}
}