Introduction to RxJS in Angular: Observables, Operators, BehaviorSubject and the async Pipe

Version note: The code and commands in this post are for Angular 12.

When Data Doesn't Arrive All at Once

You call a function, it returns a value. That is how we all start programming. But in a web app, most data doesn't arrive like that. The user clicks, types, scrolls; a timer fires every second; a response comes back from the server whenever it feels like it. These are values spread over time.

Angular uses the RxJS library to work with this kind of data. An Angular 12 project created with ng new comes with RxJS 6.6 installed, and all the code in this post targets that version.

Think of a newspaper subscription. You subscribe once and the paper lands on your doorstep every morning. It keeps coming until you cancel or the paper shuts down. The observable at the heart of RxJS is exactly that: a stream of values over time that you subscribe to.

Observable vs Promise

When JavaScript developers hear "async," they think of Promise. Let's put the two side by side:

import { Observable } from 'rxjs';

const promise = new Promise<number>(resolve => {
  console.log('Promise çalıştı');
  resolve(1);
});

const numbers$ = new Observable<number>(subscriber => {
  console.log('Observable çalıştı');
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});

numbers$.subscribe(value => console.log('Gelen değer:', value));

Run this and "Promise çalıştı" shows up right away, even though nobody called then on the Promise. The observable, on the other hand, does nothing until subscribe is called. When someone subscribes, it runs and sends three values in order.

There are three key differences. A Promise runs the moment it is created; an observable runs when you subscribe. A Promise produces a single value; an observable produces as many as it likes. A Promise cannot be canceled once it starts; you can leave an observable whenever you want.

The $ at the end of the variable name is not a rule, just a common habit. It signals to the reader "this is an observable."

subscribe and unsubscribe

Instead of a single function, you can also pass subscribe an object that handles three cases separately:

import { interval, Subscription } from 'rxjs';

const subscription: Subscription = interval(1000).subscribe({
  next: value => console.log(value),
  error: err => console.error(err),
  complete: () => console.log('Bitti')
});

setTimeout(() => subscription.unsubscribe(), 4500);

next runs on every new value, error when something goes wrong, and complete when the stream ends. Once a stream errors or completes, it sends no more values.

subscribe returns a Subscription. That is your membership card. Call unsubscribe() and the subscription is canceled. In the example, interval sends a number every second; after 0, 1, 2 and 3 arrive, we unsubscribe at four and a half seconds. complete never runs, because interval doesn't end on its own; we walked away.

Functions That Create Observables

You don't write new Observable every time. RxJS comes with ready-made creators:

import { from, fromEvent, interval, of } from 'rxjs';

of('elma', 'armut', 'kiraz').subscribe(fruit => console.log(fruit));

from([10, 20, 30]).subscribe(n => console.log(n));
from(Promise.resolve(42)).subscribe(n => console.log(n));

interval(1000).subscribe(i => console.log('saniye', i));

fromEvent(document, 'click').subscribe(() => console.log('tıklandı'));

of sends the values you give it in order and completes. from turns an array or a Promise into an observable. interval produces numbers starting from 0 at the interval you specify and never stops. fromEvent turns a DOM element's events into a stream.

There is a critical distinction here: of and from finish their job and complete. interval and fromEvent are infinite. Subscribe to an infinite stream and forget to let go, and the memory leak we will look at shortly is waiting at the door.

pipe and Operators

You rarely use a stream as is. Usually you filter it, transform it, or slip something in between. That is what pipe and operators are for:

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

interval(1000).pipe(
  tap(n => console.log('ham değer:', n)),
  filter(n => n % 2 === 0),
  map(n => n * 10),
  take(3)
).subscribe(value => console.log('sonuç:', value));

Think of operators as a factory assembly line. Each value moves along the belt from left to right and goes through a step at every station. tap looks at the value without touching it; perfect for logging. filter lets only even numbers through. map multiplies what gets through by ten. take(3) completes the stream after three values. In the console you see the results 0, 20 and 40, and then the stream closes by itself.

Pay attention to the import lines. In RxJS 6, which ships with Angular 12, operators come from 'rxjs/operators', while creators like of and interval and the Observable class come from 'rxjs'. In newer examples online you may see operators imported from 'rxjs' as well; that arrived with RxJS 7.2.

A Real Example: A Search Box

Where operators really shine is user input. Let's build a box that searches products as the user types. First, a service that pretends to be the server:

import { Injectable } from '@angular/core';
import { Observable, of } from 'rxjs';
import { delay } from 'rxjs/operators';

@Injectable({ providedIn: 'root' })
export class ProductService {
  private products = ['Mekanik Klavye', 'Kablosuz Mouse', 'Monitör Standı', 'USB-C Hub'];

  search(term: string): Observable<string[]> {
    const query = term.toLocaleLowerCase('tr-TR');
    const matches = this.products.filter(p => p.toLocaleLowerCase('tr-TR').includes(query));
    return of(matches).pipe(delay(300));
  }
}

delay(300) holds the response back by 300 milliseconds to mimic a real network request. Now the component:

import { Component, OnInit } from '@angular/core';
import { Observable, Subject } from 'rxjs';
import { debounceTime, distinctUntilChanged, switchMap } from 'rxjs/operators';
import { ProductService } from '../product.service';

@Component({
  selector: 'app-product-search',
  template: `
    <input #box (input)="search(box.value)" placeholder="Ürün ara...">
    <p *ngFor="let name of results$ | async">{{ name }}</p>
  `
})
export class ProductSearchComponent implements OnInit {
  results$!: Observable<string[]>;
  private searchTerms = new Subject<string>();

  constructor(private productService: ProductService) { }

  search(term: string): void {
    this.searchTerms.next(term);
  }

  ngOnInit(): void {
    this.results$ = this.searchTerms.pipe(
      debounceTime(300),
      distinctUntilChanged(),
      switchMap(term => this.productService.search(term))
    );
  }
}

On every keystroke, we pass the input's value to the search method using the template reference variable we saw earlier, and it pushes the value into the searchTerms stream. We will unpack Subject in a moment; for now, read it as "an observable you can push values into by hand."

debounceTime(300) works like an elevator door. As long as people keep getting in, the door keeps waiting, and it closes 300 milliseconds after the last person. While the user types "klavye", we search once when they stop typing, not once per letter.

distinctUntilChanged() drops a value that is the same as the previous one. If the user deletes a letter and types the same letter back, the same search doesn't run again.

switchMap switches to the service's search for every new term, and if the previous search is still in flight, it cancels it. That way, a slow old response can't overwrite the result of the new search. We will see how much this matters with real HTTP requests in upcoming posts.

State in a Service With Subject and BehaviorSubject

Let's go back to the cart service we saw earlier. The badge showed the count, but there was no way to tell anyone when the cart changed. Now we turn the cart into a stream:

import { Injectable } from '@angular/core';
import { BehaviorSubject, Observable } from 'rxjs';
import { map } from 'rxjs/operators';

export interface CartItem {
  name: string;
  price: number;
}

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

  add(item: CartItem): void {
    this.itemsSubject.next([...this.itemsSubject.value, item]);
  }

  clear(): void {
    this.itemsSubject.next([]);
  }
}

A Subject is an observable you can both subscribe to and push values into from outside with next(). But it has no memory; a late subscriber misses the earlier values. Like a live radio broadcast: you hear it from the moment you tune in.

A BehaviorSubject is created with an initial value and remembers the latest one. When a new subscriber arrives, it hands over the current value immediately. That is why it is ideal for holding state: no matter when the badge is created, it gets the cart's current state right away. If you want to read the current value synchronously, there is the .value property.

itemsSubject is private; we expose only a read-only stream to the outside with asObservable(). That way no component can bypass the service and call next on the cart. Inside add, we create a new array instead of modifying the old one; the reference logic we saw with pure pipes applies here too.

The async Pipe: Let Angular Handle the Subscription

The badge component now looks like this:

@Component({
  selector: 'app-cart-badge',
  template: `
    <span>Sepet: {{ count$ | async }} ürün</span>
    <div *ngIf="items$ | async as items">
      <p *ngFor="let item of items">{{ item.name }}</p>
    </div>
  `
})
export class CartBadgeComponent {
  count$ = this.cartService.count$;
  items$ = this.cartService.items$;

  constructor(private cartService: CartService) { }
}

There is no subscribe anywhere in the component. The async pipe subscribes to the observable from the template, prints every new value, and cancels the subscription itself when the component is destroyed. Zero chance of forgetting.

The *ngIf="items$ | async as items" pattern is used all the time. It binds the value from the stream to a template variable named items, so you don't subscribe to the same observable twice. Wherever you can, prefer the async pipe over a manual subscribe.

Memory Leaks and the takeUntil Pattern

Sometimes you really do need to subscribe inside a component, for example to show a notification every time the cart changes. Then you have to close the subscription yourself. In the Angular 12 world, the most common way to do that is this pattern:

import { Component, OnDestroy, OnInit } from '@angular/core';
import { Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';
import { CartService } from '../cart.service';

@Component({
  selector: 'app-cart-notifier',
  template: `<p>{{ message }}</p>`
})
export class CartNotifierComponent implements OnInit, OnDestroy {
  message = '';
  private destroy$ = new Subject<void>();

  constructor(private cartService: CartService) { }

  ngOnInit(): void {
    this.cartService.count$
      .pipe(takeUntil(this.destroy$))
      .subscribe(count => this.message = `Sepette ${count} ürün var`);
  }

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

takeUntil(this.destroy$) completes the stream the moment destroy$ emits a value. In the ngOnDestroy we saw earlier, we send a signal through destroy$, and every subscription in this component closes at once. If a component has five subscriptions, adding takeUntil(this.destroy$) to all five is enough. Always put takeUntil last inside pipe; some operators that come after it can keep the subscription alive.

What happens if you skip this? count$ is built on a BehaviorSubject, and since the service is a singleton, it never completes. Even after the component leaves the screen, the subscription lives on, and on every cart change the code of a component that no longer exists runs. Every time you open the component, one more subscription piles up. Later Angular versions brought tools that shorten this pattern; we will see them in the version posts.

A Small Experiment

Write a MouseTrackerComponent. In ngOnInit, subscribe to fromEvent<MouseEvent>(document, 'mousemove'), use map to keep only clientX and clientY, log them with tap, and show the coordinates on screen. Toggle the component on and off from AppComponent with a button and *ngIf.

Try it without takeUntil first: after closing the component, move the mouse and you will see the console keep logging. Then add the destroy$ pattern and watch the difference. As a last step, remove the manual subscribe entirely and show the coordinates with the async pipe.

In this post we faked every "server" response with of and delay. To send requests to a real API and get the response back as an observable, catch errors, and attach a token to every request, Angular has its own tool. Next up, we start talking to the server with HttpClient.