Angular: RxJS for Reactive Programming

Level: Intermediate
Recommended prerequisite: Angular Signals (separate course)

Table of Contents

1. Reactive Programming with RxJS and Signals
   • Why Reactive Programming?
   • RxJS and Signals for Reactive Programming
   • Why is RxJS Important in Angular?
   • Observable, Observer, Subscription
   • When to Use Observable vs Signal?
2. Reacting to Events
   • Use Case: Reacting to Keyboard Events
   • The Demo Application
   • Creating an Observable
   • Subscribing to an Observable
3. RxJS Operators and Pipeline
   • RxJS Operators
   • Transforming Emissions with map
   • Filtering Emissions with filter
   • Unsubscribing from an Observable
4. Retrieving Data into a Signal Using an Observable
   • HttpClient vs Resource API
   • Calling rxResource()
   • Use Case: Retrieving Data into a Signal
   • Transforming Retrieved Data with a Pipeline
5. Reactively Retrieving Data
   • Use Case: Reactive Retrieval with rxResource()
   • Working with Complex Data Structures
   • Use Case: Using RxJS Operators to Retrieve Data
   • Displaying Resource Value Signal Data
   • When to Use rxResource()?
6. Debouncing and Transforming User Input
   • Use Case: Search Feature
   • debounceTime() and distinctUntilChanged() Operators
   • Input Transformation
   • When to Use RxJS in Angular?
7. RxJS Operator Reference Tables
8. Summary and Best Practices

1. Reactive Programming with RxJS and Signals

Why Reactive Programming?

Reactive programming means writing code that observes events (user actions, state changes) and reacts accordingly. It improves the responsiveness and interactivity of applications, enhancing the user experience.

┌──────────────────────────────────────────────────────────────────┐
│              Reactive Programming Model                           │
│                                                                    │
│   SOURCE         OBSERVE          NOTIFY           REACT          │
│  ────────       ──────────       ──────────       ──────────      │
│  Event      →   Observable   →   Subscriber  →   UI/Action       │
│  (keypress,     (listens)         (receives)       (update)        │
│   HTTP, etc.)                                                      │
└──────────────────────────────────────────────────────────────────┘

Two pillars of reactive programming in Angular:

RxJS and Signals for Reactive Programming

RxJS (Reactive Extensions for JavaScript) is a popular library for reactive programming. Its key concepts:

• You observe events and data streams over time
• You subscribe to receive a notification when the event occurs
• You react to that notification by performing an operation

flowchart LR
    A[Source\nkeyboard / HTTP / timer] -->|emits data| B[Observable]
    B -->|subscribe| C[Observer / Subscriber]
    C -->|reacts| D[Action\nUI update]
    C -->|unsubscribe| E[Stop notifications\nprevent memory leaks]
    style B fill:#1976D2,color:#fff
    style C fill:#388E3C,color:#fff

Signals in Angular (since v16):

• Data value + change notification
• Recommended for defining and managing state/data in Angular components
• Improve change detection: Angular knows precisely what changed

// Signal for state
const selectedProduct = signal<Product | null>(null);
const quantity = signal<number>(1);

// Derived signal (computed)
const total = computed(() => (selectedProduct()?.price ?? 0) * quantity());

Why is RxJS Important in Angular?

Even with signals and httpResource, RxJS remains indispensable in Angular because it is integrated into several core features:

mindmap
  root((RxJS in Angular))
    Router
      Route parameters
      Route data
      Navigation events
    Reactive Forms
      Input changes
      Async validations
    HttpClient
      Returns an Observable
      Response handling
    Operators
      Data composition
      Transformation
      Filtering
      Debouncing

Concrete example: keyboard events

  Keyboard         Observable          Component
     │                  │                  │
     │── keydown ──────>│                  │
     │                  │── subscribe ─────│
     │                  │<─── key: 'R' ────│
     │                  │── transform ─>   │
     │                  │  (uppercase 'R') │
     │                  │── emit 'R' ──────>│
     │                  │                  │── moveRight()
     │── keydown ──────>│                  │
     │                  │── emit 'L' ──────>│
     │                  │                  │── moveLeft()
     │                  │<─── unsubscribe ─│

Observable, Observer, Subscription

Conveyor Belt Analogy:

  ┌─────────────────────────────────────────────────────────────┐
  │  Source (apples)                                            │
  │      │                                                      │
  │      ▼                                                      │
  │  ┌────────┐  next()   ┌─────────┐  next()   ┌──────────┐  │
  │  │ Apple 🍎│ ────────> │ Clean   │ ────────> │  Label   │  │
  │  │ emitted│           │ (filter)│           │  (map)   │  │
  │  └────────┘           └─────────┘           └──────────┘  │
  │      │                                            │        │
  │  error() ──> handle error, conveyor stops         │        │
  │  complete() ──────────────────────────────────────┘        │
  │                                          Observer reacts   │
  └─────────────────────────────────────────────────────────────┘

The Observer receives 3 types of notifications:

Basic syntax:

import { Observable, Subscription } from 'rxjs';

// Convention: $ suffix for observables
const fruits$ = new Observable<Fruit>(/* ... */);

// Subscribe → starts the stream
const sub: Subscription = fruits$.subscribe({
  next: (fruit) => console.log('New item:', fruit),
  error: (err) => console.error('Error:', err),
  complete: () => console.log('No more items'),
});

// Unsubscribe → stops the stream, prevents memory leaks
sub.unsubscribe();

Important convention: The $ suffix on a variable indicates it is an Observable. E.g.: helpKey$, searchText$, products$.

When to Use Observable vs Signal?

flowchart TD
    Q{What is the need?}
    Q -->|Manage state\nor data| S[Signal ✅\nconst data = signal(initialValue)]
    Q -->|Listen to DOM events\nor async data over time| O[Observable ✅\nfromEvent / rxResource]
    Q -->|Derived value from signals| C[Computed Signal ✅\nconst derived = computed()]
    Q -->|Simple HTTP request| R[httpResource ✅\nsimpler, less code]
    Q -->|HTTP + complex pipeline\nor forkJoin| RX[rxResource + RxJS operators ✅]

Mnemonic rule:

“When working with state, a signal is great. For an event-driven stream, an observable is keen.”

2. Reacting to Events

Use Case: Reacting to Keyboard Events

Scenario: Implement global keyboard shortcuts (hotkeys):

• ? → display a help message
• Escape → hide the help message

Why RxJS is the right choice for events?

• Events arrive asynchronously and at unpredictable intervals
• There can be a stream of events (not a single value)
• We want to filter specific events
• We can define an operator pipeline to transform the event before reacting

Implementation steps:

1. Write the help message content (HTML)
2. Declare state (showHelp signal)
3. Create an Observable that listens to keyboard events
4. Subscribe to the Observable
5. Transform the event → key value (map)
6. Filter → only ? and Escape (filter)
7. React → update showHelp (tap)

The Demo Application

The APM (Acme Product Management) application is used throughout the course. It includes:

src/app/
├── app.component.ts          ← Main component
├── app.config.ts             ← Configuration (HttpClient, etc.)
├── product-selection/
│   ├── product-selection.component.ts    ← Product component
│   └── product-selection.component.html
├── product/
│   ├── product.service.ts    ← Service + shared state
│   └── product.model.ts
├── review/
│   ├── review-list.component.ts
│   ├── review-search.component.ts
│   └── review.service.ts
└── supplier/
    └── supplier.service.ts

Creating an Observable

RxJS provides several observable creation functions:

Implementation in the component:

import { Component, signal, OnInit } from '@angular/core';
import { fromEvent, Observable } from 'rxjs';

@Component({
  selector: 'app-product-selection',
  templateUrl: './product-selection.component.html',
})
export class ProductSelectionComponent implements OnInit {
  // State with Signal
  showHelp = signal(false);

  // Observable ($ convention)
  private helpKey$!: Observable<KeyboardEvent>;

  ngOnInit() {
    // Create the Observable: listens to all keydown events on the document
    this.helpKey$ = fromEvent<KeyboardEvent>(document, 'keydown');
  }
}

fromEvent flow diagram:

  document (DOM)
      │
      │ keydown event
      ▼
  ┌─────────────────────────────────────────┐
  │  fromEvent(document, 'keydown')          │
  │                                         │
  │  Observable<KeyboardEvent>              │
  │  Emits on each key press                │
  └──────────────────┬──────────────────────┘
                     │ subscribe()
                     ▼
              Observer / Component
              (receives each event)

Subscribing to an Observable

The Observable emits nothing until subscribed to.

// Subscribe: starts receiving events
const sub = this.helpKey$.subscribe((event: KeyboardEvent) => {
  console.log('Key pressed:', event.key);
  this.showHelp.set(true);
});

ASCII Marble Diagram — Keyboard event stream:

  Time ──────────────────────────────────────────────>

  keydown$:
  ──a──────b──────?──────n──────Esc────────────────────
    │       │      │      │      │
    │       │      │      │      └── Escape emitted
    │       │      │      └──────── 'n' emitted
    │       │      └─────────────── '?' emitted
    │       └────────────────────── 'b' emitted
    └────────────────────────────── 'a' emitted

3. RxJS Operators and Pipeline

RxJS Operators

Pipeable operators are chained in a pipeline via the .pipe() method. They transform, filter, and compose items before they are emitted to the subscriber.

flowchart LR
    O[Observable\nsource] -->|pipe| P1[map\ntransform]
    P1 -->|pipe| P2[filter\ncriterion]
    P2 -->|pipe| P3[tap\nside effect]
    P3 -->|subscribe| S[Subscriber\ncomponent]
    style O fill:#1565C0,color:#fff
    style P1 fill:#1976D2,color:#fff
    style P2 fill:#1976D2,color:#fff
    style P3 fill:#1976D2,color:#fff
    style S fill:#2E7D32,color:#fff

General pipeline syntax:

const result$ = source$.pipe(
  operator1(/* arguments */),
  operator2(/* arguments */),
  operator3(/* arguments */),
);

result$.subscribe(value => console.log(value));

Concrete example with of and operators:

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

of(1, 2, 3, 4, 5).pipe(
  tap(n => console.log('Before map:', n)),      // side effect, no modification
  map(n => n * 2),                              // transforms: 2, 4, 6, 8, 10
  filter(n => n > 5),                           // filters: 6, 8, 10
  tap(n => console.log('After filter:', n)),    // logs: 6, 8, 10
).subscribe(n => console.log('Result:', n));
// Output: 6, 8, 10

ASCII Marble Diagram — map and filter:

  source$:    ─1─────2─────3─────4─────5──|
                │     │     │     │     │
  map(*2):    ─2─────4─────6─────8────10──|
                │     │     │     │     │
  filter(>5): ───────────────6─────8────10──|
                              │     │     │
  subscribe:                  6     8    10

Transforming Emissions with map

The map operator transforms each emitted element according to a function.

import { fromEvent } from 'rxjs';
import { map, tap, filter } from 'rxjs/operators';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

// In the component
private initKeyboardObservable() {
  fromEvent<KeyboardEvent>(document, 'keydown').pipe(
    // 1. Transform the event into the key name (string)
    map((event: KeyboardEvent) => event.key),
    
    // 2. tap for debugging (does NOT affect the emitted value)
    tap(key => console.log('Key pressed:', key)),
    
    // 3. Filter: only ? and Escape
    filter(key => key === '?' || key === 'Escape'),
    
    // 4. React without modifying the value
    tap(key => this.showHelp.set(key === '?')),
    
    // 5. Automatic unsubscribe when component is destroyed
    takeUntilDestroyed(),
  ).subscribe();
}

The role of tap:

  Incoming value     tap(fn)    Outgoing value
  ─────────────────────────────────────────────
       'a'      ──→  log('a')  ──→    'a'
       '?'      ──→  log('?')  ──→    '?'
       Esc      ──→  log(Esc)  ──→    Esc
  
  tap does NOT modify the value, it performs a side effect

Tip: tap is your best friend for debugging an RxJS pipeline. Add it first if your pipeline is not producing the expected results.

Filtering Emissions with filter

// filter: only elements satisfying the condition pass through
filter(key => key === '?' || key === 'Escape')

ASCII Marble Diagram — Complete keydown pipeline:

  fromEvent$: ──a────b────?────n────Esc──────────────>
                │    │    │    │     │
  map(e.key): ──'a'──'b'──'?'──'n'──'Esc'────────────>
                │    │    │    │     │
  tap(log):   ──'a'──'b'──'?'──'n'──'Esc'────────────>
  (logs all)
                          │          │
  filter(?/Esc):──────────'?'────────'Esc'────────────>
                          │          │
  tap(set):    ──────────set(true)──set(false)─────────>
                          │          │
  subscribe:  ──────────(nothing)───(nothing)──────────>
  (no logic needed here, everything is in tap)

Unsubscribing from an Observable

Two approaches for unsubscribe:

Option 1: Manual with OnDestroy

import { Component, OnDestroy, OnInit, signal } from '@angular/core';
import { fromEvent, Subscription } from 'rxjs';

@Component({ /* ... */ })
export class ProductSelectionComponent implements OnInit, OnDestroy {
  showHelp = signal(false);
  private keyboardSub!: Subscription;

  ngOnInit() {
    this.keyboardSub = fromEvent<KeyboardEvent>(document, 'keydown').pipe(
      // ... operators
    ).subscribe();
  }

  ngOnDestroy() {
    // Stops notifications, prevents memory leaks
    this.keyboardSub.unsubscribe();
  }
}

Option 2: Automatic with takeUntilDestroyed (recommended)

import { Component, signal } from '@angular/core';
import { fromEvent } from 'rxjs';
import { map, filter, tap } from 'rxjs/operators';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

@Component({ /* ... */ })
export class ProductSelectionComponent {
  showHelp = signal(false);

  constructor() {
    // IMPORTANT: must be in an injection context (constructor or class property)
    fromEvent<KeyboardEvent>(document, 'keydown').pipe(
      map(event => event.key),
      filter(key => key === '?' || key === 'Escape'),
      tap(key => this.showHelp.set(key === '?')),
      takeUntilDestroyed(), // ← Angular operator, from '@angular/core/rxjs-interop'
    ).subscribe();
  }
}

Important: takeUntilDestroyed must be used in an injection context (constructor or class property declaration). It is not a standard RxJS operator, but an Angular operator.

4. Retrieving Data into a Signal Using an Observable

HttpClient vs Resource API

Before Angular 19 — Complex flow with HttpClient:

1. Define the HTTP request
2. subscribe() to trigger the request
3. Wait for the response (asynchronous)
4. Put data into a Signal manually
5. unsubscribe() → don't forget!

Since Angular 19 — Simplified Resource API:

flowchart LR
    R[rxResource\nor httpResource] -->|declares| REQ[HTTP Request]
    REQ -->|auto-subscribe| HTTP[HttpClient.get]
    HTTP -->|response| SIG[value Signal\nauto-populated]
    SIG -->|bind| UI[Template / UI]
    R --> ERR[error Signal]
    R --> LOAD[isLoading Signal]
    style R fill:#E65100,color:#fff
    style SIG fill:#2E7D32,color:#fff

Comparison of the three approaches:

Calling rxResource()

import { rxResource } from '@angular/core/rxjs-interop';
import { inject } from '@angular/core';
import { HttpClient } from '@angular/common/http';

// Simple form (no parameters)
productsResource = rxResource({
  stream: () => this.http.get<Product[]>(this.url),
  defaultValue: [],  // avoids handling undefined everywhere
});

// Reactive form (with signal-dependent parameters)
reviewsResource = rxResource({
  params: () => this.productService.selectedProduct(),  // dependent signal
  stream: (p) => {
    const product = p.params;
    if (!product) return undefined; // skip if no product
    return this.http.get<Review[]>(`${this.reviewUrl}?productId=${product.id}`);
  },
  defaultValue: [],
});

Properties returned by rxResource:

Use Case: Retrieving Data into a Signal

Required configuration in app.config.ts:

import { ApplicationConfig } from '@angular/core';
import { provideHttpClient } from '@angular/common/http';

export const appConfig: ApplicationConfig = {
  providers: [
    provideHttpClient(),
    // other providers...
  ]
};

Product service with rxResource:

import { Injectable, inject } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { rxResource } from '@angular/core/rxjs-interop';
import { map } from 'rxjs/operators';
import { Product } from './product.model';

@Injectable({ providedIn: 'root' })
export class ProductService {
  private http = inject(HttpClient);
  private productsUrl = 'api/products';

  // Resource declaration — the request is sent automatically
  productsResource = rxResource({
    stream: () => this.http.get<Product[]>(this.productsUrl).pipe(
      // Observable pipeline to transform data
      map(products => products.sort((a, b) => 
        a.productName.localeCompare(b.productName)
      ))
    ),
    defaultValue: [] as Product[],
  });

  // State signal: selected product
  selectedProduct = signal<Product | null>(null);
}

Transforming Retrieved Data with a Pipeline

// In the rxResource stream, operators can be chained
stream: () => this.http.get<Product[]>(this.url).pipe(
  // Sort by name
  map(products => [...products].sort((a, b) => 
    a.productName.localeCompare(b.productName)
  )),
  // Filter active products
  map(products => products.filter(p => p.isActive)),
  // Debug log
  tap(products => console.log('Products loaded:', products.length)),
),

rxResource flow diagram:

  rxResource declared
        │
        ▼
  stream() called automatically
        │
        ▼
  http.get<Product[]>(url)
        │   (Observable)
        ▼
  .pipe(
    map(sort),       ← transformation
    tap(log),        ← side effect
  )
        │
        ▼ (server response)
  value Signal ← data automatically injected
        │
        ▼
  Template displays data

5. Reactively Retrieving Data

Use Case: Reactive Retrieval with rxResource()

When using params in rxResource, the stream automatically re-executes on every change of the dependent signal.

// In review.service.ts
reviewsResource = rxResource({
  // params: dependent signal → triggers stream re-execution
  params: () => this.productService.selectedProduct(),
  stream: (p) => {
    const product = p.params;
    if (!product) return undefined; // avoids an unnecessary request
    return this.http.get<Review[]>(
      `${this.reviewUrl}?productId=${product.id}`
    );
  },
  defaultValue: [] as Review[],
});

Signal → rxResource reactivity diagram:

  User selects product A
         │
         ▼
  selectedProduct signal changes → "Product A"
         │
         ▼
  rxResource detects the change (params)
         │
         ▼
  stream() re-executed with params = "Product A"
         │
         ▼
  New HTTP request: GET /reviews?productId=A
         │
         ▼
  value Signal updated with reviews for A
         │
         ▼
  UI automatically re-rendered

  (same if user selects product B)

Working with Complex Data Structures

RxJS excels at composing complex data. Use case: display the suppliers for a product, given that the product only contains supplierIds[].

The Problem:

Product {
  id: 1,
  name: "Hammer",
  supplierIds: [10, 20, 30]   ← IDs only, not the names!
}

Need: retrieve each supplier by ID
→ 3 parallel HTTP requests
→ Combine the results

Solution with forkJoin:

flowchart TD
    A[supplierIds: 10, 20, 30] --> B[forkJoin]
    B --> C1[GET /suppliers/10]
    B --> C2[GET /suppliers/20]
    B --> C3[GET /suppliers/30]
    C1 --> D[Wait for ALL responses]
    C2 --> D
    C3 --> D
    D --> E[Emit array\nof complete Suppliers]
    style B fill:#7B1FA2,color:#fff
    style D fill:#7B1FA2,color:#fff

combineLatest for two parallel sources:

import { combineLatest } from 'rxjs';
import { map } from 'rxjs/operators';

// Retrieve products AND categories in parallel,
// then map category IDs to names
const productsWithCategories$ = combineLatest([
  this.http.get<Product[]>('/api/products'),
  this.http.get<Category[]>('/api/categories'),
]).pipe(
  map(([products, categories]) =>
    products.map(product => ({
      ...product,
      categoryName: categories.find(c => c.id === product.categoryId)?.name
    }))
  )
);

ASCII Marble Diagram — combineLatest:

  products$:   ────────────[P1,P2,P3]──────────────────>
  categories$: ──────[C1,C2,C3]───────────────────────>
                                      │
  combineLatest:──────────────────────[P1,P2,P3]+[C1,C2,C3]→
                                      │
  map(merge):  ──────────────────────[P1+CatName, P2+CatName, ...]→

Use Case: Using RxJS Operators to Retrieve Data

forkJoin — Multiple Parallel Requests:

import { forkJoin } from 'rxjs';
import { map } from 'rxjs/operators';

// In supplier.service.ts
@Injectable({ providedIn: 'root' })
export class SupplierService {
  private http = inject(HttpClient);
  private productService = inject(ProductService);

  // Computed signal: IDs of suppliers for the selected product
  private supplierIds = computed(() =>
    this.productService.selectedProduct()?.supplierIds
  );

  // Reactive resource: retrieves suppliers when IDs change
  suppliersResource = rxResource({
    params: () => {
      const ids = this.supplierIds();
      return ids?.length ? ids : undefined; // undefined = skip
    },
    stream: (p) => {
      const supplierIds = p.params as number[];
      // forkJoin waits for ALL responses before emitting
      return forkJoin(
        supplierIds.map(id =>
          this.http.get<Supplier>(`/api/suppliers/${id}`)
        )
      );
    },
    defaultValue: [] as Supplier[],
  });
}

forkJoin vs combineLatest:

• forkJoin: waits for all Observables to complete, then emits a single array — ideal for parallel HTTP requests
• combineLatest: emits as soon as all have emitted at least one value, then on each new emission — ideal for continuous streams

Displaying Resource Value Signal Data

// In product-selection.component.ts
@Component({ /* ... */ })
export class ProductSelectionComponent {
  private supplierService = inject(SupplierService);

  // Supplier signal from the resource
  selectedProductSuppliers = this.supplierService.suppliersResource.value;

  // Computed: transforms the array into a readable string
  suppliers = computed(() =>
    this.selectedProductSuppliers()
      .map(s => s.supplierName)
      .join(', ')
  );
}

<!-- product-selection.component.html -->
<div>
  <strong>Suppliers:</strong> {{ suppliers() }}
</div>

When to Use rxResource()?

6. Debouncing and Transforming User Input

Use Case: Search Feature

Problem: A search that sends an HTTP request on every keystroke → inefficient and costly.

User types:      "s" "se" "sea" "sear" "searc" "search"
                   │   │    │     │      │       │
Without debounce: req req  req   req    req     req  (6 requests!)
With debounce:                                  req  (1 request after pause)

The Signal challenge: Signals are synchronous — they don’t understand the concept of time. This is where RxJS comes in.

Solution: toObservable + RxJS pipeline:

import { toObservable } from '@angular/core/rxjs-interop';
import { debounceTime, distinctUntilChanged, map } from 'rxjs/operators';
import { rxResource } from '@angular/core/rxjs-interop';

@Injectable({ providedIn: 'root' })
export class ReviewService {
  private http = inject(HttpClient);

  // Signal bound to the search input
  enteredSearch = signal('');

  // Signal → Observable conversion to enable debouncing
  private searchText$ = toObservable(this.enteredSearch).pipe(
    debounceTime(400),              // wait 400ms of inactivity
    distinctUntilChanged(),         // ignore if the value hasn't changed
    map(text => text.toLowerCase()), // transform to lowercase
    tap(text => console.log('searchText:', text)), // debug
  );

  // Signal derived from the debounced searchText$
  private searchText = toSignal(this.searchText$, { initialValue: '' });

  // rxResource reactive on the debounced signal
  reviewsResource = rxResource({
    params: () => this.searchText(),
    stream: (p) => this.http.get<Review[]>(
      `${this.reviewUrl}?search=${p.params}`
    ).pipe(
      map(reviews => reviews.sort((a, b) => a.title.localeCompare(b.title)))
    ),
    defaultValue: [] as Review[],
  });
}

debounceTime() and distinctUntilChanged() Operators

How debounceTime(400) works:

  Keystrokes: ─s──e──a──r──c──h─────────────>
               │  │  │  │  │  │
  Timer:         400ms  400ms  400ms  400ms  400ms  400ms→ FIRE!
               restart  restart  restart...

  After 400ms of silence:
  debounce$:  ─────────────────────"search"──>
                                        │
  HTTP request emitted only once ───────┘

  Marble diagram debounceTime(400ms):

  source$:    ─s───e──a──r────────c──h────────────────>
                                  ↑ pause > 400ms
  debounce$:  ────────────"sear"────────────"search"──>

How distinctUntilChanged() works:

  source$:     ─"cat"──"cat"──"cats"──"cats"──"dog"──>
  distinct$:   ─"cat"─────────"cats"────────────"dog"──>
                        ↑ duplicate ignored  ↑ duplicate ignored

Complete search pipeline flow:

flowchart LR
    I[User input\nenteredSearch Signal] -->|toObservable| O[Observable]
    O --> D[debounceTime\n400ms]
    D --> U[distinctUntilChanged]
    U --> M[map\nlowerCase]
    M --> T[tap\nconsole.log]
    T -->|toSignal| S[Signal\nsearchText]
    S --> R[rxResource\nparams: searchText]
    R --> H[HTTP GET\n/reviews?search=...]
    H --> V[value Signal]
    V --> UI[Template]
    style D fill:#F57F17,color:#fff
    style U fill:#F57F17,color:#fff
    style R fill:#E65100,color:#fff

Input Transformation

You can also transform retrieved data with a pipeline in the rxResource stream:

reviewsResource = rxResource({
  params: () => this.searchText(),
  stream: (p) => this.http.get<Review[]>(
    `${this.reviewUrl}?search=${p.params}`
  ).pipe(
    // Filter reviews without a title
    filter(reviews => reviews.length > 0),
    // Sort by title
    map(reviews => reviews.sort((a, b) => a.title.localeCompare(b.title))),
    // Debug log
    tap(reviews => console.log('Reviews found:', reviews.length)),
  ),
  defaultValue: [] as Review[],
});

Complete pipeline (input → transformation → display):

  Input signal ──→ toObservable ──→ debounceTime(400)
       │                                    │
       │                          distinctUntilChanged()
       │                                    │
       │                              map(toLowerCase)
       │                                    │
       │                              toSignal ──→ searchText signal
       │                                    │
       └────────────────────────────────────┘
                                            │
                                     rxResource(params: searchText)
                                            │
                                     http.get(url + searchText)
                                            │
                                     .pipe(map(sort), tap(log))
                                            │
                                     value Signal ──→ Template

When to Use RxJS in Angular?

7. RxJS Operator Reference Tables

Transformation Operators

Filtering Operators

Combination Operators (Creation Functions)

Utility Operators

Signal ↔ Observable Conversion Operators (Angular)

8. Summary and Best Practices

Modern Angular Reactive Architecture

flowchart TB
    subgraph Template["Template (HTML)"]
        UI[Display\nbindings, events]
    end

    subgraph Component["Component (.ts)"]
        SIG[Signals\nlocal state]
        COMP[Computed\nderived values]
        EFF[Effects\nside effects]
    end

    subgraph Service["Service (.ts)"]
        OBS[Observables\nRxJS]
        RES[Resources\nrxResource / httpResource]
        STATE[Shared Signals\nglobal state]
    end

    subgraph Backend["Backend / API"]
        HTTP[HTTP Server]
        WS[WebSocket]
        DOM[DOM Events]
    end

    DOM -->|fromEvent| OBS
    HTTP -->|HttpClient.get| RES
    WS -->|fromEvent| OBS
    OBS -->|toSignal| STATE
    RES --> STATE
    STATE --> SIG
    SIG --> COMP
    COMP --> UI
    SIG --> EFF
    EFF --> HTTP

    style Template fill:#1A237E,color:#fff
    style Component fill:#0D47A1,color:#fff
    style Service fill:#01579B,color:#fff
    style Backend fill:#006064,color:#fff

Best Practices

1. Naming Convention

// ✅ $ suffix for Observables
const keydown$ = fromEvent(document, 'keydown');
const searchText$ = toObservable(searchSignal);

// ✅ No $ for Signals
const products = signal<Product[]>([]);
const selectedProduct = signal<Product | null>(null);

2. Always Unsubscribe

// ✅ Prefer takeUntilDestroyed (less boilerplate)
fromEvent(document, 'keydown').pipe(
  takeUntilDestroyed(),
).subscribe();

// ✅ Or use rxResource/httpResource (automatic unsubscribe)
productsResource = rxResource({
  stream: () => this.http.get<Product[]>(this.url),
});

3. Debugging with tap

// ✅ Add tap to inspect the pipeline
someObservable$.pipe(
  tap(v => console.log('Step 1:', v)),  // before transformation
  map(transformFn),
  tap(v => console.log('Step 2:', v)),  // after transformation
  filter(filterFn),
  tap(v => console.log('Step 3:', v)),  // after filter
).subscribe();

4. Choosing the Right Mapping Operator

// switchMap: cancels the previous request (search, navigation)
searchText$.pipe(
  switchMap(text => http.get(`/api/search?q=${text}`))
)

// mergeMap: parallel execution (multiple downloads)
ids$.pipe(
  mergeMap(id => http.get(`/api/items/${id}`))
)

// concatMap: sequential (ordered processing)
actions$.pipe(
  concatMap(action => processAction(action))
)

// exhaustMap: ignores if already in progress (form submission)
submitBtn$.pipe(
  exhaustMap(() => http.post('/api/submit', data))
)

5. Error Handling

// ✅ Always handle errors in an HTTP pipeline
this.http.get<Product[]>(this.url).pipe(
  catchError(err => {
    console.error('HTTP Error:', err);
    return of([]); // Return a replacement Observable
  })
)

Quick Decision Flow

Need to react to...?
├── Data/state → Signal ✅
├── Derived value → computed() ✅
├── Side effect → effect() ✅
├── DOM event (click, key, scroll) → fromEvent() + RxJS ✅
├── Simple HTTP GET → httpResource ✅
├── HTTP GET + pipeline/composition → rxResource ✅
├── HTTP POST/PUT/DELETE → Direct HttpClient ✅
├── Debounce/throttle → RxJS debounceTime/throttleTime ✅
├── Multiple parallel sources → combineLatest / forkJoin ✅
└── Complex nested data → RxJS operators ✅

Useful Resources and Links

Course notes generated from the content of the “Angular: RxJS for Reactive Programming” course by Deborah Kurata.

Search Terms

angular · rxjs · reactive · programming · frontend · development · operators · data · observable · case · signal · retrieving · rxresource · transforming · emissions · events · filtering · input · operator · option · pipeline · reacting · resource · signals