Advanced RxJS and State Management

RxJS Operators in Depth

RxJS is the backbone of Angular's async programming. At the intermediate level, you used catchError and retry. At the advanced level, you need operators that handle real-world complexity — debouncing user input, cancelling stale requests, and combining multiple streams.

switchMap — cancel the previous request when a new one comes in:

This is the most important operator for search boxes. Without it, if a user types quickly, you can get responses arriving out of order.

// search.component.ts

import { Component, OnInit } from '@angular/core';

import { FormControl } from '@angular/forms';

import { CourseService } from './course.service';

import { switchMap, debounceTime, distinctUntilChanged } from 'rxjs/operators';

@Component({

selector: 'app-search',

template: `

<input [formControl]="searchControl" placeholder="Search courses...">

<ul>

<li *ngFor="let course of results">{{ course.name }}</li>

</ul>

`

})

export class SearchComponent implements OnInit {

searchControl = new FormControl('');

results: any[] = [];

constructor(private courseService: CourseService) {}

ngOnInit(): void {

this.searchControl.valueChanges.pipe(

debounceTime(300), // wait 300ms after the user stops typing

distinctUntilChanged(), // ignore if the value hasn't changed

switchMap(term => // cancel previous request, fire new one

this.courseService.search(term || '')

)

).subscribe(results => this.results = results);

}

}

combineLatest — react when any of multiple streams emits:

import { combineLatest } from 'rxjs';

// Fires whenever either the filter or the sort order changes

combineLatest([this.filter$, this.sortOrder$]).pipe(

switchMap(([filter, sort]) => this.courseService.getCourses(filter, sort))

).subscribe(courses => this.courses = courses);

**forkJoin — wait for multiple requests to all complete:**

import { forkJoin } from 'rxjs';

// Load courses and user profile at the same time, proceed when both are done

forkJoin({

courses: this.courseService.getCourses(),

profile: this.userService.getProfile()

}).subscribe(({ courses, profile }) => {

this.courses = courses;

this.profile = profile;

});

NgRx State Management

When your app grows beyond a few components, passing data around through services and @Input()/@Output() chains becomes hard to track. NgRx gives you a single, predictable store for all application state — modelled on Redux.

The four core pieces are: Store (holds the state), Actions (describe what happened), Reducers (update the state), and Selectors (read from the state).

Installing NgRx:

ng add @ngrx/store

ng add @ngrx/effects

ng add @ngrx/store-devtools

Defining actions:

// course.actions.ts

import { createAction, props } from '@ngrx/store';

import { Course } from './course.model';

export const loadCourses = createAction('[Course List] Load Courses');

export const loadCoursesSuccess = createAction(

'[Course List] Load Courses Success',

props<{ courses: Course[] }>()

);

export const loadCoursesFailure = createAction(

'[Course List] Load Courses Failure',

props<{ error: string }>()

);

Writing the reducer:

// course.reducer.ts

import { createReducer, on } from '@ngrx/store';

import { loadCourses, loadCoursesSuccess, loadCoursesFailure } from './course.actions';

import { Course } from './course.model';

export interface CourseState {

courses: Course[];

loading: boolean;

error: string | null;

}

const initialState: CourseState = {

courses: [],

loading: false,

error: null

};

export const courseReducer = createReducer(

initialState,

on(loadCourses, state => ({ ...state, loading: true, error: null })),

on(loadCoursesSuccess, (state, { courses }) => ({ ...state, courses, loading: false })),

on(loadCoursesFailure, (state, { error }) => ({ ...state, error, loading: false }))

);

Handling side effects with NgRx Effects:

Effects are where your API calls live. They listen for an action, call the API, then dispatch a success or failure action.

// course.effects.ts

import { Injectable } from '@angular/core';

import { Actions, createEffect, ofType } from '@ngrx/effects';

import { CourseService } from './course.service';

import { loadCourses, loadCoursesSuccess, loadCoursesFailure } from './course.actions';

import { switchMap, map, catchError } from 'rxjs/operators';

import { of } from 'rxjs';

@Injectable()

export class CourseEffects {

loadCourses$ = createEffect(() =>

this.actions$.pipe(

ofType(loadCourses),

switchMap(() =>

this.courseService.getCourses().pipe(

map(courses => loadCoursesSuccess({ courses })),

catchError(error => of(loadCoursesFailure({ error: error.message })))

)

)

)

);

constructor(private actions$: Actions, private courseService: CourseService) {}

}

Reading state in a component using Selectors:

// course-list.component.ts

import { Store } from '@ngrx/store';

import { loadCourses } from './course.actions';

import { selectAllCourses, selectLoading } from './course.selectors';

export class CourseListComponent implements OnInit {

courses$ = this.store.select(selectAllCourses);

loading$ = this.store.select(selectLoading);

constructor(private store: Store) {}

ngOnInit(): void {

this.store.dispatch(loadCourses());

}

}

<div *ngIf="loading$ | async">Loading...</div>

<ul>

<li *ngFor="let course of courses$ | async">{{ course.name }}</li>

</ul>