# RxJS Fundamentals Course: [Angular](https://www.devtutorials.ai/learn/angular) · Section: Working with RxJS Source: https://www.devtutorials.ai/learn/angular/rxjs/rxjs-fundamentals > 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. An **Observable** is the main type in RxJS. It sends values one after another, at the moment they happen. The next sections show how to create one and how to read its values. You may know a **Promise** from JavaScript. This table compares a Promise and an Observable: | Point | Promise | Observable | |---|---|---| | Number of values | One | Zero, one or many | | When it starts its work | At once, when you create it | Only when some code subscribes to it | | Can you cancel it? | No | Yes, with `unsubscribe()` | | How to change the values | `.then()` | Operators in `pipe()` | ## 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. ```angular runnable title="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) {

} `, }) export class App { log: string[] = []; constructor() { const orderStatus$ = new Observable(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: ```ts 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. ```ts 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.

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. ```ts 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. ```angular runnable title="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) {

} `, }) 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"**. ```angular runnable title="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 { const delay = term === 'lap' ? 900 : 300; return timer(delay).pipe(map(() => `Results for "${term}"`)); } @Component({ selector: 'app-root', imports: [AsyncPipe], template: `

`, }) export class App { private terms$ = new Subject(); 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. | Operator | What it does | Example use | |---|---|---| | `map()` | Changes each value | Change server data into text for the screen | | `filter()` | Drops the values that fail a check | Ignore an empty search text | | `tap()` | Runs extra code; the value does not change | Write a log message, show a loading spinner | | `switchMap()` | Starts a new Observable and cancels the previous one | Search box, load data for a route | ## 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`.

A BehaviorSubject shares one current value with every subscriber.

In the next example, both subscribers subscribe **after** the value `'Asha'` is sent. ```angular runnable title="Subject vs BehaviorSubject" import { Component } from '@angular/core'; import { Subject, BehaviorSubject } from 'rxjs'; @Component({ selector: 'app-root', template: ` @for (line of log; track $index) {

} `, }) export class App { log: string[] = []; constructor() { const subject$ = new Subject(); const behavior$ = new BehaviorSubject('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. | Point | Subject | BehaviorSubject | |---|---|---| | Starting value | None | Required | | A new subscriber receives | Only values sent later | The stored value at once, then values sent later | | Read the stored value | Not possible | `.value` | | Example use | Events: a click, a "saved" message | Data: the signed-in user, the cart | In a service, keep the Subject `private` and give other classes `subject.asObservable()`. Other classes can then subscribe, but only the service can send values with `next()`. ## 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. ```angular runnable title="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([]); 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: `

Items in cart: {{ cart.count | async }}

`, }) 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. ```angular runnable title="catchError and finalize" import { Component } from '@angular/core'; import { Observable, catchError, finalize, of, throwError } from 'rxjs'; function loadOrders(fail: boolean): Observable { return fail ? throwError(() => new Error('Server error 500')) : of(['#1001', '#1002']); } @Component({ selector: 'app-root', template: `

Orders: {{ orders.join(', ') || 'none' }}

Message: {{ message }}

Loading: {{ loading }}

`, }) 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`. The place of `catchError()` matters. Inside `switchMap()`, it replaces the error of one request with a fallback value, and the search box keeps working. On the outer Observable, the whole Observable stops after the first error, so later searches do nothing. ## 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()`: ```ts 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. ```angular runnable title="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: `

Live orders panel is open. It checks for new orders every second.

`, }) 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: ` @if (open) { } @for (line of log.lines; track $index) {

} `, }) 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)`. | Way to stop a subscription | Code you write | Best for | |---|---|---| | `async` pipe | No extra code | Showing values in the template | | `takeUntilDestroyed()` | One operator | `subscribe()` in a component or service | | `unsubscribe()` in `ngOnDestroy()` | A field and a method | Older code | ## Common Mistakes

| Mistake | What you see | Fix | |---|---|---| | Creating the `pipe()` but never subscribing | Nothing happens, and no request is sent | Call `subscribe()`, or use the `async` pipe | | `subscribe()` to `interval()` without stopping it | The timer keeps running after the component is removed | Add `takeUntilDestroyed()` | | Calling `subscribe()` inside another `subscribe()` | Old requests are not cancelled, and the code is hard to read | Use `switchMap()` | | Using `Subject` for data such as the current user | A subscriber that subscribes late receives nothing | Use `BehaviorSubject` | | Using `tap()` to change values | The values do not change | Use `map()` to change values | | No error handling | One error stops the Observable, and the screen shows nothing | Add `catchError()` |

## 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`.