RxJS Fundamentals

By Syncfusion®, Inc. · Last updated November 5, 2026

Working with values over time using RxJS: Observables, observers, subscriptions, the pipe, map, filter, tap and switchMap operators, Subjects, BehaviorSubject, the async pipe, error handling and automatic cleanup with takeUntilDestroyed.

Introduction to RxJS

Some values in an app do not arrive at once. They arrive later, or they arrive many times. Examples:

• A server sends a response after some time.
• A user presses keys one by one.
• A timer sends a number every second.
• The route parameter changes when the user opens another product.

RxJS (Reactive Extensions for JavaScript) is a library for working with these values. Every Angular project includes it. Angular itself uses it: the HTTP client returns server responses with RxJS, and ActivatedRoute.paramMap from the routing lesson also uses RxJS.

Real-World Example

A product search page sends a request to the server each time the user changes the search text. The user types "lap" and then "laptop". The server answers "lap" slowly, so its results arrive last and replace the correct "laptop" results.

With RxJS, the app cancels the "lap" request as soon as the user searches for "laptop". The page always displays the results for the newest search text.

You may know a Promise from JavaScript. This table compares a Promise and an Observable:

Observables

An Observable is an object that sends values over time. To get the values, you subscribe to it with the subscribe() method.

An Observable sends three types of messages:

• next(value) sends a new value. It can send many values.
• error(err) reports a failure, for example a server error. After an error, the Observable sends nothing more.
• complete() reports that there are no more values. After complete, the Observable sends nothing more.

Create an Observable

You can create an Observable with new Observable(...). You rarely do this in real apps, but it shows clearly how an Observable works.

In real apps, you usually use RxJS functions that create an Observable for you:

• of(1, 2, 3) sends the values 1, 2 and 3.
• from([...]) sends each item of an array, one by one.
• interval(1000) sends 0, 1, 2 and so on, one number every second.

Creating and subscribing to an Observable

import { Component } from '@angular/core';
import { Observable } from 'rxjs';

@Component({
  selector: 'app-root',
  template: `
    @for (line of log; track $index) {
      <p>{{ line }}</p>
    }
  `,
})
export class App {
  log: string[] = [];

  constructor() {
    const orderStatus$ = new Observable<string>(subscriber => {
      subscriber.next('Order received');
      subscriber.next('Packed');
      subscriber.next('Shipped');
      subscriber.complete();
    });

    this.write('Before subscribe');
    orderStatus$.subscribe({
      next: status => this.write(`Status: ${status}`),
      complete: () => this.write('No more updates'),
    });
  }

  write(line: string) {
    this.log.push(line);
  }
}

The first line is "Before subscribe". The Observable sends its values only after the code calls subscribe().

The name orderStatus$ ends with $. Developers add $ to the end of a variable name to show that it stores an Observable.

Observers

An observer is the object that receives the messages of an Observable. It has up to three functions, one for each message type:

const observer = {
  next: (value: number) => console.log('Value:', value),
  error: (err: unknown) => console.error('Failed:', err),
  complete: () => console.log('Done'),
};

of(1, 2, 3).subscribe(observer);

• You can leave out any of the three functions.
• If you need only the values, pass one function: source$.subscribe(value => ...). This function is used for next.

Subscriptions

subscribe() returns a Subscription object. Call its unsubscribe() method to stop getting values. This also stops the work of the Observable, for example a timer or a server request.

const sub = interval(1000).subscribe(n => console.log(n));

// later, for example when the user leaves the page:
sub.unsubscribe();

Which Observables Need unsubscribe()?

• Observables that complete, such as of(...) and a server request, stop by themselves. You do not need to unsubscribe.
• Observables that never complete, such as interval(), run forever. You must unsubscribe. The last two sections of this lesson show how.

RxJS Operators

An operator is a function that changes the values of an Observable. It takes an Observable and returns a new Observable with the changed values. The original Observable does not change.

1, 2, 3, 4Observablefilter(n%2==0)2, 4map(n*10)20, 40subscribereceives 20, 40Values flow left to right. Each operator returns a new Observable.

An Observable piped through operators.

pipe()

pipe() connects operators in a row. Each value goes through the operators one by one, from left to right.

source$.pipe(operatorA(), operatorB(), operatorC());

map()

map() changes each value into a new value. For example, map(n => n * 10) changes 2 into 20. It works like map() on arrays.

filter()

filter() checks each value with a function. It sends the value on only when the function returns true. Other values are dropped.

tap()

tap() runs extra code for each value, for example writing a log message. It does not change the value. The same value goes to the next operator.

The next example uses all three operators on a list of orders.

pipe with filter, map and tap

import { Component } from '@angular/core';
import { from, filter, map, tap } from 'rxjs';

interface Order {
  id: number;
  total: number;
  paid: boolean;
}

const ORDERS: Order[] = [
  { id: 1, total: 120, paid: true },
  { id: 2, total: 80, paid: false },
  { id: 3, total: 45, paid: true },
];

@Component({
  selector: 'app-root',
  template: `
    @for (line of log; track $index) {
      <p>{{ line }}</p>
    }
  `,
})
export class App {
  log: string[] = [];

  constructor() {
    from(ORDERS)
      .pipe(
        tap(order => this.write(`checking order ${order.id}`)),
        filter(order => order.paid),
        map(order => `Order ${order.id}: $${order.total} paid`),
      )
      .subscribe(text => this.write(text));
  }

  write(line: string) {
    this.log.push(line);
  }
}

What happens to each order:

1. tap() writes "checking order ..." for every order.
2. filter() keeps only the paid orders. Order 2 is not paid, so it stops here.
3. map() changes each paid order into a line of text.

switchMap()

switchMap() changes each value into a new Observable, for example a server request. When a new value arrives, switchMap() cancels the previous request and starts a new one.

Use it for search boxes, and for loading data when a route parameter changes.

In the next example, the request for "lap" takes 900 ms and the request for "laptop" takes 300 ms. Click Search "lap", then quickly click Search "laptop".

switchMap keeps only the latest request

import { Component } from '@angular/core';
import { AsyncPipe } from '@angular/common';
import { Subject, Observable, switchMap, map, timer, startWith } from 'rxjs';

// Simulates a server: answers after a delay that depends on the term.
function searchApi(term: string): Observable<string> {
  const delay = term === 'lap' ? 900 : 300;
  return timer(delay).pipe(map(() => `Results for "${term}"`));
}

@Component({
  selector: 'app-root',
  imports: [AsyncPipe],
  template: `
    <button (click)="search('lap')">Search "lap"</button>
    <button (click)="search('laptop')">Search "laptop"</button>
    <p>{{ result$ | async }}</p>
  `,
})
export class App {
  private terms$ = new Subject<string>();

  result$ = this.terms$.pipe(
    switchMap(term => searchApi(term)),
    startWith('Click both buttons quickly, left first.'),
  );

  search(term: string) {
    this.terms$.next(term);
  }
}

startWith() gives the first text to display. The async pipe in the template displays each new value. You learn it in the section "Async Pipe" below.

The page displays the results for "laptop". switchMap() cancelled the slow "lap" request, so its results never replace the "laptop" results.

Subjects

A Subject is an Observable that you can send values into. You send a value with next(), and every subscriber receives it.

Services use a Subject to tell many components about an event, for example "the cart changed".

• A Subject does not store a value. A subscriber receives only the values that are sent after it subscribes.
• Call complete() when the Subject will not send more values.

BehaviorSubject

A BehaviorSubject is a Subject that stores its newest value.

• You must give it a starting value, for example new BehaviorSubject('Guest').
• When a new subscriber subscribes, it receives the stored value at once.
• You can read the stored value at any time with .value.

BehaviorSubjectcurrent: 'Asha'Headergets 'Asha' nowProfilegets 'Asha' nowLate subscriberstill gets 'Asha'Every subscriber receives the current value at once, then each new value.

A BehaviorSubject shares one current value with every subscriber.

In the next example, both subscribers subscribe after the value 'Asha' is sent.

Subject vs BehaviorSubject

import { Component } from '@angular/core';
import { Subject, BehaviorSubject } from 'rxjs';

@Component({
  selector: 'app-root',
  template: `
    @for (line of log; track $index) {
      <p>{{ line }}</p>
    }
  `,
})
export class App {
  log: string[] = [];

  constructor() {
    const subject$ = new Subject<string>();
    const behavior$ = new BehaviorSubject<string>('Guest');

    subject$.next('Asha');
    behavior$.next('Asha');

    // Both subscribe after the value 'Asha' was sent.
    subject$.subscribe(name => this.write(`Subject gave: ${name}`));
    behavior$.subscribe(name => this.write(`BehaviorSubject gave: ${name}`));

    subject$.next('Ravi');
    behavior$.next('Ravi');
    this.write(`behavior$.value is ${behavior$.value}`);
  }

  write(line: string) {
    this.log.push(line);
  }
}

The Subject's subscriber did not receive 'Asha', because the Subject does not store values. The BehaviorSubject's subscriber received 'Asha' at once, because the BehaviorSubject stored it.

Async Pipe

The async pipe subscribes to an Observable from the template. It displays the newest value. When Angular removes the component, the pipe unsubscribes by itself.

So you do not write subscribe() or unsubscribe() for this Observable.

A cart service shown with the async pipe

import { Component, Injectable, inject } from '@angular/core';
import { AsyncPipe } from '@angular/common';
import { BehaviorSubject, map } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class CartService {
  private items$ = new BehaviorSubject<string[]>([]);
  readonly items = this.items$.asObservable();
  readonly count = this.items$.pipe(map(list => list.length));

  add(name: string) {
    this.items$.next([...this.items$.value, name]);
  }
}

@Component({
  selector: 'app-root',
  imports: [AsyncPipe],
  template: `
    <button (click)="cart.add('Mouse')">Add mouse</button>
    <button (click)="cart.add('Cable')">Add cable</button>
    <p>Items in cart: {{ cart.count | async }}</p>
    <p>{{ (cart.items | async)?.join(', ') }}</p>
  `,
})
export class App {
  cart = inject(CartService);
}

Two rules for the async pipe:

• Before the first value arrives, the pipe returns null. So write ?. when you read a property of the value, for example (cart.items | async)?.join(', ').
• Each async pipe creates its own subscription. Two async pipes on one server request send the request two times.

RxJS Error Handling

When an Observable sends an error, it stops and sends no more values. If the subscriber has no error function, the error is not handled, and the user sees nothing useful. Two operators help with errors.

catchError()

catchError() runs when an error happens. It returns a replacement Observable, for example an empty list. The subscriber then receives the replacement value instead of the error.

finalize()

finalize() runs a function when the Observable stops for any reason: it completes, it has an error, or the code unsubscribes. Use it to hide a loading spinner.

catchError and finalize

import { Component } from '@angular/core';
import { Observable, catchError, finalize, of, throwError } from 'rxjs';

function loadOrders(fail: boolean): Observable<string[]> {
  return fail ? throwError(() => new Error('Server error 500')) : of(['#1001', '#1002']);
}

@Component({
  selector: 'app-root',
  template: `
    <p>Orders: {{ orders.join(', ') || 'none' }}</p>
    <p>Message: {{ message }}</p>
    <p>Loading: {{ loading }}</p>
  `,
})
export class App {
  orders: string[] = [];
  message = '';
  loading = true;

  constructor() {
    loadOrders(true)
      .pipe(
        catchError(err => {
          this.message = `Could not load orders (${err.message})`;
          return of([]);
        }),
        finalize(() => (this.loading = false)),
      )
      .subscribe(list => (this.orders = list));
  }
}

loadOrders(true) sends an error. catchError() writes a message and returns an empty list. finalize() sets loading to false.

Change loadOrders(true) to loadOrders(false) and click Run. The orders appear, and finalize() again sets loading to false.

Unsubscribing from Observables

Some Observables never complete, for example interval(). If you subscribe to one in a component and Angular removes the component, the subscription keeps running:

• The function in subscribe() keeps running.
• The timer or the connection stays open.
• The browser cannot free the memory.

This problem is called a memory leak. The app becomes slower over time.

Unsubscribe in ngOnDestroy()

One way to fix this is to store the Subscription and call unsubscribe() in ngOnDestroy():

export class Ticker implements OnDestroy {
  private sub = interval(1000).subscribe(n => console.log('Tick', n));

  ngOnDestroy() {
    this.sub.unsubscribe();
  }
}

When You Must Unsubscribe

• You do not need to unsubscribe when you use the async pipe, or when the Observable completes (a server request, of, from).
• You must unsubscribe when you call subscribe() yourself on interval(), a Subject, paramMap, or browser events from fromEvent().

Automatic Subscription Cleanup

takeUntilDestroyed()

takeUntilDestroyed() is an operator from @angular/core/rxjs-interop. When Angular removes the component, this operator completes the Observable. So the subscription stops by itself.

Add it as the last operator in pipe(). You do not need ngOnDestroy().

In the next example, the panel uses interval() to check for new orders every second. Click Toggle panel to remove the panel.

takeUntilDestroyed stops the timer

import { Component, Injectable, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { interval, finalize } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class EventLog {
  lines: string[] = [];
  add(line: string) {
    this.lines.push(line);
  }
}

@Component({
  selector: 'app-live-orders',
  template: `<p>Live orders panel is open. It checks for new orders every second.</p>`,
})
export class LiveOrders {
  refreshes = 0;
  private log = inject(EventLog);

  constructor() {
    this.log.add('Panel opened, timer started');
    interval(1000)
      .pipe(
        takeUntilDestroyed(),
        finalize(() => this.log.add('Panel closed, timer stopped')),
      )
      .subscribe(() => this.refreshes++);
  }
}

@Component({
  selector: 'app-root',
  imports: [LiveOrders],
  template: `
    <button (click)="open = !open">Toggle panel</button>
    @if (open) {
      <app-live-orders />
    }
    @for (line of log.lines; track $index) {
      <p>{{ line }}</p>
    }
  `,
})
export class App {
  open = true;
  log = inject(EventLog);
}

When the panel is removed, takeUntilDestroyed() stops the timer, and finalize() writes "Panel closed, timer stopped". Without takeUntilDestroyed(), the timer would keep running.

Where to Call It

• Without an argument, call it only in the constructor or in a field, the same as inject().
• In other places, pass a DestroyRef object: write destroyRef = inject(DestroyRef) in a field, then takeUntilDestroyed(this.destroyRef).

Common Mistakes

Check Yourself

Exercises

What to Remember

• An Observable sends values over time. It starts when code calls subscribe(), and it stops on complete, error or unsubscribe().
• pipe() connects operators: map() changes values, filter() drops values, tap() runs extra code, and switchMap() starts a new request and cancels the previous one. catchError() replaces an error with a fallback value, and finalize() runs when the Observable stops.
• Use BehaviorSubject for data that many components share. Use the async pipe in templates. Add takeUntilDestroyed() to every subscribe() on an Observable that never completes.

Next: API Integration, how to get data from a server and send data to it with HttpClient.