CodeWorlds

Angular course · Module 7: RxJS

RxJS Operators - Ninja Techniques

2 min read ·
In this lesson3

Operators are pure functions that transform streams - like different ninja techniques changing the way information is passed.

Transformation Operators

import { map, switchMap, mergeMap, concatMap, exhaustMap } from 'rxjs/operators';

// map - transforms each element
source$.pipe(
  map(value => value * 2)
);

// switchMap - switches to a new stream, cancels the previous one
// Ideal for search (cancel previous query)
searchInput$.pipe(
  switchMap(query => this.http.get(`/search?q=${query}`))
);

// mergeMap - parallel execution of all
// Good when order doesn't matter
ids$.pipe(
  mergeMap(id => this.http.get(`/item/${id}`))
);

// concatMap - sequential, preserves order
// When order is important
actions$.pipe(
  concatMap(action => this.saveAction(action))
);

// exhaustMap - ignores new ones until the current one completes
// Ideal for save buttons (preventing double clicks)
saveButton$.pipe(
  exhaustMap(() => this.saveData())
);

Filtering Operators

import { filter, take, takeUntil, takeWhile, distinctUntilChanged, debounceTime } from 'rxjs/operators';

// filter - passes only matching elements
numbers$.pipe(
  filter(n => n > 5)
);

// take - takes the first n elements
numbers$.pipe(
  take(3)  // Only the first 3
);

// distinctUntilChanged - removes consecutive duplicates
values$.pipe(
  distinctUntilChanged()  // [1,1,2,2,3,1] -> [1,2,3,1]
);

// debounceTime - waits for a pause in emissions
searchInput$.pipe(
  debounceTime(300)  // Waits 300ms without new values
);

Combination Operators

import { combineLatest, forkJoin, merge, concat, zip } from 'rxjs';

// combineLatest - emits when any emits (after first emissions)
combineLatest([user$, settings$]).subscribe(([user, settings]) => {
  console.log(user, settings);
});

// forkJoin - waits for all to complete (like Promise.all)
forkJoin([
  this.http.get('/api/samurai'),
  this.http.get('/api/weapons')
]).subscribe(([samurai, weapons]) => {
  // Both completed
});

// merge - combines streams in parallel
merge(clicks$, keyups$).subscribe(event => {
  console.log('Event:', event);
});
Code for this lesson: src/app/app.component.ts
import { Component, signal, OnDestroy } from '@angular/core';
import { CommonModule } from '@angular/common';
import { Observable, Subject, of, from, timer, combineLatest, forkJoin } from 'rxjs';
import { map, delay, takeUntil, mergeMap, switchMap, concatMap, exhaustMap, catchError } from 'rxjs/operators';

@Component({
  selector: 'app-root',
  standalone: true,
  imports: [CommonModule],
  template: `
    <div class="operators-dojo">
      <h1>RxJS Operators Dojo</h1>

      <section class="transformation">
        <h2>Transformation (map, mergeMap, switchMap)</h2>

        <div class="demo-box">
          <h3>Map - value transformation</h3>
          <button (click)="demoMap()">Run Map</button>
          <div class="result">{{ mapResult() }}</div>
        </div>

        <div class="demo-box">
          <h3>MergeMap - all in parallel</h3>
          <button (click)="demoMergeMap()">Run MergeMap</button>
          <div class="result">
            @for (item of mergeMapResults(); track $index) {
              <span class="tag">{{ item }}</span>
            }
          </div>
        </div>

        <div class="demo-box">
          <h3>SwitchMap - cancel previous</h3>
          <button (click)="demoSwitchMap()">Click quickly!</button>
          <div class="result">
            Clicks: {{ switchClickCount() }}<br>
            Result: {{ switchMapResult() }}
          </div>
        </div>
      </section>

      <section class="combination">
        <h2>Combination (combineLatest, forkJoin)</h2>

        <div class="demo-box">
          <h3>CombineLatest - latest from each</h3>
          <div class="sliders">
            <label>
              Strength: {{ strength() }}
              <input type="range" min="1" max="100" [value]="strength()"
                     (input)="onStrengthChange($event)">
            </label>
            <label>
              Agility: {{ agility() }}
              <input type="range" min="1" max="100" [value]="agility()"
                     (input)="onAgilityChange($event)">
            </label>
          </div>
          <div class="result">
            Total power: {{ combinedPower() }}
          </div>
        </div>

        <div class="demo-box">
          <h3>ForkJoin - wait for all</h3>
          <button (click)="demoForkJoin()" [disabled]="forkJoinLoading()">
            {{ forkJoinLoading() ? 'Loading...' : 'Fetch all data' }}
          </button>
          <div class="result">
            @if (forkJoinResult()) {
              <div>Samurai: {{ forkJoinResult()?.samurai }}</div>
              <div>Weapons: {{ forkJoinResult()?.weapons }}</div>
              <div>Missions: {{ forkJoinResult()?.missions }}</div>
            }
          </div>
        </div>
      </section>

      <section class="error-handling">
        <h2>Error Handling (catchError)</h2>

        <div class="demo-box">
          <h3>CatchError - defense against errors</h3>
          <button (click)="demoWithError()">Simulate API error</button>
          <div class="result" [class.error]="hasError()">
            {{ errorResult() }}
          </div>
        </div>
      </section>
    </div>
  `,
  styles: [`
    .operators-dojo {
      font-family: 'Segoe UI', sans-serif;
      padding: 20px;
      background: linear-gradient(135deg, #2d1b4e 0%, #1a1a2e 100%);
      min-height: 100vh;
      color: #eee;
    }

    h1 {
      text-align: center;
      color: #a855f7;
    }

    h2 {
      background: #a855f7;
      color: white;
      padding: 10px;
      border-radius: 5px;
    }

    section {
      margin: 20px 0;
    }

    .demo-box {
      background: rgba(255,255,255,0.1);
      padding: 15px;
      border-radius: 10px;
      margin: 10px 0;
    }

    .demo-box h3 {
      margin-top: 0;
      color: #a855f7;
    }

    button {
      padding: 10px 20px;
      background: #a855f7;
      color: white;
      border: none;
      border-radius: 5px;
      cursor: pointer;
      margin: 5px 0;
    }

    button:hover:not(:disabled) {
      background: #c084fc;
    }

    button:disabled {
      opacity: 0.5;
      cursor: not-allowed;
    }

    .result {
      background: rgba(0,0,0,0.3);
      padding: 10px;
      border-radius: 5px;
      margin-top: 10px;
      min-height: 30px;
    }

    .result.error {
      border: 2px solid #ef4444;
      color: #fca5a5;
    }

    .tag {
      display: inline-block;
      background: #a855f7;
      padding: 3px 8px;
      border-radius: 3px;
      margin: 2px;
      font-size: 0.9em;
    }

    .sliders {
      display: flex;
      flex-direction: column;
      gap: 10px;
    }

    .sliders label {
      display: flex;
      align-items: center;
      gap: 10px;
    }

    .sliders input[type="range"] {
      flex: 1;
    }
  `]
})
export class AppComponent implements OnDestroy {
  private destroy$ = new Subject<void>();
  private switchClick$ = new Subject<void>();

  // Map demo
  mapResult = signal('');

  // MergeMap demo
  mergeMapResults = signal<string[]>([]);

  // SwitchMap demo
  switchClickCount = signal(0);
  switchMapResult = signal('');

  // CombineLatest demo
  strength = signal(50);
  agility = signal(50);
  combinedPower = signal(100);

  // ForkJoin demo
  forkJoinLoading = signal(false);
  forkJoinResult = signal<any>(null);

  // Error handling demo
  hasError = signal(false);
  errorResult = signal('Click the button to test');

  constructor() {
    // SwitchMap - cancels previous request
    this.switchClick$.pipe(
      takeUntil(this.destroy$),
      switchMap(() => {
        const clickNum = this.switchClickCount();
        return timer(2000).pipe(
          map(() => `Completed for click #${clickNum}`)
        );
      })
    ).subscribe(result => this.switchMapResult.set(result));
  }

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

  demoMap(): void {
    const samurai = ['Musashi', 'Kojiro', 'Jubei'];

    from(samurai).pipe(
      map(name => `Samurai: ${name.toUpperCase()}`)
    ).subscribe({
      next: val => this.mapResult.set(val),
      complete: () => this.mapResult.update(v => v + '')
    });
  }

  demoMergeMap(): void {
    this.mergeMapResults.set([]);

    const missions = ['Recon', 'Defense', 'Attack'];

    from(missions).pipe(
      mergeMap(mission =>
        of(`${mission} completed`).pipe(
          delay(Math.random() * 2000)
        )
      )
    ).subscribe(result => {
      this.mergeMapResults.update(arr => [...arr, result]);
    });
  }

  demoSwitchMap(): void {
    this.switchClickCount.update(c => c + 1);
    this.switchMapResult.set('Processing...');
    this.switchClick$.next();
  }

  onStrengthChange(event: Event): void {
    const value = +(event.target as HTMLInputElement).value;
    this.strength.set(value);
    this.updateCombinedPower();
  }

  onAgilityChange(event: Event): void {
    const value = +(event.target as HTMLInputElement).value;
    this.agility.set(value);
    this.updateCombinedPower();
  }

  private updateCombinedPower(): void {
    this.combinedPower.set(this.strength() + this.agility());
  }

  demoForkJoin(): void {
    this.forkJoinLoading.set(true);
    this.forkJoinResult.set(null);

    forkJoin({
      samurai: of(5).pipe(delay(1000)),
      weapons: of(12).pipe(delay(1500)),
      missions: of(3).pipe(delay(800))
    }).subscribe(result => {
      this.forkJoinResult.set(result);
      this.forkJoinLoading.set(false);
    });
  }

  demoWithError(): void {
    this.hasError.set(false);
    this.errorResult.set('Sending request...');

    const failingApi$ = new Observable(subscriber => {
      setTimeout(() => {
        subscriber.error(new Error('Server unavailable!'));
      }, 1000);
    });

    failingApi$.pipe(
      catchError(error => {
        this.hasError.set(true);
        return of(`Error caught: ${error.message} - using fallback data`);
      })
    ).subscribe(result => this.errorResult.set(result as string));
  }
}

Spotted a mistake in this lesson?

Check yourself

Answer the questions from this lesson. Pick an answer to see right away whether it is correct.

  1. 1. What is the main difference between switchMap and mergeMap?

  2. 2. What does the map() operator do in RxJS?

These are 2 of 5 questions for this lesson. Solve the rest in the game.

Hands-on tasks in the game

  • Horizontal ordering

    Arrange the import of transformation operators from the rxjs operators package:

  • Vertical ordering

    Arrange the lines of the search operator chain in the order in which we write them inside pipe():

  • Code editor

    Complete the search stream's pipe so that it lets through only phrases with at least 3 characters.

  • Click in order

    Arrange the operator call on a stream: source$.pipe(map(...))

Useful articles

We use cookies to enhance your experience on the site
Privacy Policy
OK