RxJS Dasar — Angular

RxJS (Reactive Extensions for JavaScript) adalah library untuk pemrograman reaktif dengan Observable. Observable Dasar import { Observable, of, from } from…

RxJS adalah library untuk mengolah data yang datang bertahap dari waktu ke waktu, dan Angular memakainya di banyak tempat: HttpClient, router, form. Tipe utamanya Observable, aliran nilai yang bisa kamu dengarkan. Bayangin notifikasi status pesanan Martabak Bang Jali di aplikasi pesan-antar: diproses, dimasak, diantar. Notifikasinya baru masuk sesudah kamu menyalakan langganan (subscribe), dan berhenti saat kamu mematikannya (unsubscribe). Lupa mematikannya, ponselmu terus bergetar walau kamu sudah keluar dari halamannya.

🧪 Tebakan cepat: Sebuah komponen menulis status$ = new Observable(...) yang mencetak dapur mulai lalu mengirim tiga status. Gak ada yang memanggil subscribe(). Berapa kali dapur mulai tercetak? Jawabannya ada di bagian "Belum subscribe, belum jalan".

Observable dan subscribe

import { Observable } from 'rxjs'

const status$ = new Observable<string>((pelanggan) => {
  console.log('dapur mulai')
  pelanggan.next('diproses')
  pelanggan.next('dimasak')
  pelanggan.next('diantar')
  pelanggan.complete()
})

status$.subscribe({
  next: (s) => console.log('status:', s),
  complete: () => console.log('selesai'),
})
// dapur mulai / status: diproses / status: dimasak / status: diantar / selesai

Belum subscribe, belum jalan

Fungsi di dalam Observable baru dijalankan saat ada yang subscribe. Tanpa subscribe(), dapur mulai tercetak 0 kali dan gak ada status yang dikirim. Itu jawaban tebakannya. Observable itu malas (lazy).

Dan tiap subscribe() menjalankan fungsinya lagi dari awal:

Yang ditulis dapur mulai tercetak
status$ dibuat, gak di-subscribe 0 kali
status$.subscribe(...) sekali 1 kali, lalu 3 status
status$.subscribe(...) dua kali 2 kali, masing-masing 3 status

Ini penting di pelajaran berikutnya: request HttpClient juga Observable, jadi request-nya baru terkirim saat di-subscribe.

Membuat Observable: of, from, timer

Pembuat Yang dikirim
of(50000, 40000) 50000, lalu 40000, langsung, lalu selesai
from([4, 6, 0]) 4, 6, 0 (isi array satu per satu), lalu selesai
interval(1000) 0, 1, 2, ... tiap 1 detik, gak pernah selesai
timer(0, 2000) 0 langsung, lalu 1, 2, ... tiap 2 detik, gak pernah selesai

of dan from mengirim semuanya saat itu juga, di dalam subscribe(). interval dan timer terus jalan sampai di-unsubscribe.

pipe, map, filter

Operator dipasang di pipe() dan mengolah tiap nilai yang lewat, tanpa mengubah sumbernya:

import { from, filter, map } from 'rxjs'

from([4, 6, 0])
  .pipe(
    filter((stok) => stok > 0),
    map((stok) => stok * 10),
  )
  .subscribe((n) => console.log(n))
// 40
// 60

Status pesanan Bang Jali dibuat dengan cara yang sama. Aplikasinya bertanya ke dapur tiap 2 detik:

const URUTAN = ['diproses', 'dimasak', 'diantar']

status$ = timer(0, 2000).pipe(map((i) => URUTAN[Math.min(i, 2)]))
// detik 0: diproses, detik 2: dimasak, detik 4: diantar, detik 6: diantar, ...

Latihan memakai pola yang sama dengan satu nilai: of('angular').pipe(map((val) => val.toUpperCase())).subscribe((val) => (this.result = val)) di constructor, jadi <p>{{ result }}</p> berbunyi ANGULAR.

Unsubscribe, atau memorinya bocor

timer di atas gak pernah selesai. Kalau Pesanan subscribe sendiri lalu ditutup (misalnya App menyembunyikannya dengan @if), langganannya tetap hidup: tiap 2 detik callback-nya jalan untuk komponen yang sudah gak ada. Buka-tutup lima kali, ada lima langganan yang jalan bersamaan. Itu memory leak.

// Bocor: ngOnInit dipanggil Angular sekali sesudah komponen dibuat.
ngOnInit() {
  this.status$.subscribe((s) => (this.status = s))
}

// Aman 1: simpan langganannya, matikan di ngOnDestroy
private langganan?: Subscription
ngOnInit() {
  this.langganan = this.status$.subscribe((s) => (this.status = s))
}
ngOnDestroy() {
  this.langganan?.unsubscribe()
}

// Aman 2: takeUntilDestroyed() dari '@angular/core/rxjs-interop', ditulis di constructor
constructor() {
  this.status$.pipe(takeUntilDestroyed()).subscribe((s) => (this.status = s))
}

Satu hal lagi soal this.status = s: sejak Angular 22 komponen secara bawaan OnPush, jadi properti biasa yang diganti dari dalam subscribe gak memperbarui tampilan. Di Angular 22, simpan nilainya di signal (status = signal(''), lalu this.status.set(s)), atau biarkan template yang subscribe.

Cara aman ketiga, yang paling ringkas, adalah async pipe.

async pipe: subscribe dan unsubscribe otomatis

import { Component } from '@angular/core'
import { AsyncPipe } from '@angular/common'
import { map, timer } from 'rxjs'

const URUTAN = ['diproses', 'dimasak', 'diantar']

@Component({
  selector: 'app-pesanan',
  standalone: true,
  imports: [AsyncPipe],
  template: `<p>Status: {{ status$ | async }}</p>`,
})
export class Pesanan {
  status$ = timer(0, 2000).pipe(map((i) => URUTAN[Math.min(i, 2)]))
}

async subscribe saat komponen muncul, menampilkan tiap nilai baru (Status: diproses, lalu dimasak, lalu diantar), dan unsubscribe sendiri saat komponennya dihancurkan. Gak ada subscribe() dan gak ada yang perlu dimatikan.

📌 Subject dan BehaviorSubject

Subject adalah Observable yang nilainya kamu kirim sendiri dengan next(). Bedanya ada di pelanggan yang datang telat:

const status = new BehaviorSubject('diproses')
status.next('dimasak')
status.subscribe((s) => console.log('telat:', s)) // telat: dimasak

⚠️ Jebakan yang sering ketemu

TL;DR: Observable adalah aliran nilai dari waktu ke waktu; subscribe() mulai mendengarkan, dan tanpa subscribe gak ada yang jalan, sementara tiap subscribe menjalankannya dari awal. of dan from mengirim nilai yang sudah ada lalu selesai, interval dan timer jalan terus. pipe(filter(...), map(...)) mengolah tiap nilai. Langganan pada aliran yang gak selesai harus dihentikan saat komponen hilang: paling ringkas dengan {{ status$ | async }} (plus AsyncPipe di imports), yang subscribe dan unsubscribe otomatis, atau dengan takeUntilDestroyed(). BehaviorSubject memberikan nilai terakhirnya ke pelanggan yang datang telat; Subject biasa enggak.

Yang akan kamu pelajari