Angular'da RxJS'e Giriş: Observable, Operatörler, BehaviorSubject ve async Pipe

Sürüm notu: Bu yazıdaki kod ve komutlar Angular 12 içindir.

Veri Tek Seferde Gelmiyorsa

Bir fonksiyon çağırırsın, bir değer döner. Programlamaya böyle başlarız. Ama bir web uygulamasında verinin çoğu böyle gelmez. Kullanıcı tıklar, yazar, kaydırır; bir timer her saniye tetiklenir; sunucudan cevap ne zaman geleceği belli olmadan gelir. Bunlar zamana yayılmış değerlerdir.

Angular bu tür verilerle çalışmak için RxJS kütüphanesini kullanır. ng new ile oluşturduğun Angular 12 projesinde RxJS 6.6 kurulu gelir ve bu yazıdaki bütün kodlar o sürüme göredir.

Gazete aboneliği gibi düşün. Bir kere abone olursun, gazete her sabah kapına gelir. Sen aboneliği iptal edene ya da gazete kapanana kadar bu böyle sürer. RxJS'in merkezindeki observable tam olarak budur: abone olduğun, zamana yayılmış bir değer akışı.

Observable ve Promise Farkı

JavaScript'te asenkron iş deyince akla Promise gelir. İkisini yan yana koyalım:

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));

Bu kodu çalıştırınca "Promise çalıştı" yazısı hemen görünür; Promise'e kimse then demese bile. Observable ise subscribe çağrılana kadar hiçbir şey yapmaz. Birisi abone olduğunda çalışır ve üç değeri sırayla gönderir.

Üç temel fark var. Promise oluşturulduğu an çalışır, observable abone olunca. Promise tek bir değer döner, observable istediği kadar. Promise başladıktan sonra iptal edilemez, observable'dan ise istediğin an çıkabilirsin.

Değişken adının sonundaki $ bir kural değil, yaygın bir alışkanlık. Okuyan kişiye "bu bir observable" diye işaret eder.

subscribe ve unsubscribe

subscribe'a tek bir fonksiyon yerine üç durumu ayrı ayrı karşılayan bir nesne de verebilirsin:

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 her yeni değerde, error bir hata olduğunda, complete da akış bittiğinde çalışır. Bir akış hata verirse ya da tamamlanırsa artık değer göndermez.

subscribe geriye bir Subscription döndürür. Bu senin abonelik kartın. unsubscribe() çağırdığında abonelik iptal olur. Örnekte interval her saniye bir sayı gönderir; 0, 1, 2 ve 3 geldikten sonra, dört buçukuncu saniyede abonelikten çıkarız. complete hiç çalışmaz, çünkü interval kendi kendine bitmez; biz bıraktık.

Observable Üreten Fonksiyonlar

Her seferinde new Observable yazmazsın. RxJS hazır üreticilerle gelir:

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, verdiğin değerleri sırayla gönderip tamamlanır. from, bir diziyi ya da Promise'i observable'a çevirir. interval belirttiğin aralıkla 0'dan başlayarak sayı üretir ve hiç durmaz. fromEvent da bir DOM elemanının event'lerini akışa dönüştürür.

Burada kritik bir ayrım var: of ve from işini bitirip tamamlanır. interval ve fromEvent ise sonsuzdur. Sonsuz bir akışa abone olup bırakmayı unutursan, birazdan göreceğimiz memory leak sorunu kapıdadır.

pipe ve Operatörler

Bir akışı olduğu gibi kullanmak nadirdir. Genelde süzer, dönüştürür, araya bir şey sıkıştırırsın. Bunun için pipe ve operatörler var:

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));

Operatörlere bir fabrikanın üretim bandı gibi bak. Her değer bantta soldan sağa ilerler ve her istasyonda bir işlemden geçer. tap değere dokunmadan bakar; log yazmak için idealdir. filter sadece çift sayıları geçirir. map geçen değeri on ile çarpar. take(3) üç değer aldıktan sonra akışı tamamlar. Console'da 0, 20 ve 40 sonuçlarını görürsün, ardından akış kendiliğinden kapanır.

Import satırına dikkat et. Angular 12 ile gelen RxJS 6'da operatörler 'rxjs/operators' yolundan, of, interval gibi üreticiler ve Observable ise 'rxjs' yolundan gelir. İnternetteki yeni örneklerde operatörlerin de 'rxjs''ten import edildiğini görebilirsin; o kullanım RxJS 7.2 ile geldi.

Gerçek Bir Örnek: Arama Kutusu

Operatörlerin asıl parladığı yer kullanıcı girdisi. Kullanıcı yazdıkça ürün arayan bir kutu yapalım. Önce sunucuyu taklit eden bir service:

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) cevabı 300 milisaniye geciktirerek gerçek bir ağ isteğini taklit ediyor. Şimdi 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))
    );
  }
}

Her tuş vuruşunda daha önce gördüğümüz template reference variable ile input'un değerini search metoduna veriyoruz, o da değeri searchTerms akışına itiyor. Subject'i birazdan açacağız; şimdilik "elle değer gönderebildiğin bir observable" de.

debounceTime(300) asansör kapısı gibi çalışır. Biri bindikçe kapı beklemeye devam eder, son kişiden 300 milisaniye sonra kapanır. Kullanıcı "klavye" yazarken her harf için değil, yazmayı bıraktığında bir kere arama yaparız.

distinctUntilChanged() bir önceki değerle aynı olanı geçirmez. Kullanıcı bir harf silip aynı harfi geri yazarsa aynı arama tekrar yapılmaz.

switchMap ise her yeni terim için service'teki aramaya geçer ve önceki arama hâlâ sürüyorsa onu iptal eder. Böylece yavaş gelen eski bir cevap, yeni aramanın sonucunu ezemez. Gerçek HTTP istekleriyle bunun ne kadar önemli olduğunu ilerleyen yazılarda göreceğiz.

Subject ve BehaviorSubject ile Service'te State

Daha önce gördüğümüz sepet service'ine dönelim. Rozet sayıyı gösteriyordu ama sepet değiştiğinde kimseye haber veremiyorduk. Şimdi sepeti bir akışa çeviriyoruz:

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([]);
  }
}

Subject, hem abone olunabilen hem de next() ile dışarıdan değer itilebilen bir observable'dır. Ama hafızası yoktur; sonradan abone olan, önceki değerleri kaçırır. Canlı radyo yayını gibi: açtığın andan itibaren duyarsın.

BehaviorSubject bir başlangıç değeriyle oluşturulur ve son değeri hatırlar. Yeni abone geldiğinde ona hemen güncel değeri verir. State tutmak için bu yüzden ideal: rozet ne zaman oluşursa oluşsun, sepetin o anki hâlini anında alır. Güncel değeri senkron okumak istersen .value property'si var.

itemsSubject private, dışarıya sadece asObservable() ile salt-okunur bir akış açıyoruz. Böylece hiçbir component service'i atlayıp sepete next diyemez. add içinde de diziyi değiştirmek yerine yeni bir dizi oluşturuyoruz; pure pipe'larda gördüğümüz referans mantığı burada da geçerli.

async Pipe: Aboneliği Angular'a Bırak

Rozet component'i artık şöyle:

@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) { }
}

Component'te hiçbir subscribe yok. async pipe template'te observable'a kendisi abone olur, her yeni değeri ekrana basar ve component yok edildiğinde aboneliği kendisi iptal eder. Unutma riski sıfır.

*ngIf="items$ | async as items" kalıbı çok sık kullanılır. Akıştan gelen değeri items adıyla template'te bir değişkene bağlar, böylece aynı observable'a iki kez abone olmazsın. Mümkün olan her yerde manuel subscribe yerine async pipe'ı tercih et.

Memory Leak ve takeUntil Deseni

Bazen component içinde gerçekten subscribe etmen gerekir; örneğin sepet her değiştiğinde bir bildirim göstermek için. O zaman aboneliği kendin kapatmalısın. Angular 12 dünyasında bunun en yaygın yolu şu desen:

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$), destroy$ bir değer gönderdiği an akışı tamamlar. Daha önce gördüğümüz ngOnDestroy'da destroy$'a bir sinyal gönderiyoruz ve bu component'teki bütün abonelikler aynı anda kapanıyor. Bir component'te beş abonelik varsa beşine de takeUntil(this.destroy$) eklemen yeterli. takeUntil'i pipe içinde her zaman en sona koy; arkasına gelen bazı operatörler aboneliği canlı tutabilir.

Bunu yapmazsan ne olur? count$ bir BehaviorSubject'e dayanıyor ve service singleton olduğu için asla tamamlanmıyor. Component ekrandan gitse bile abonelik yaşar, her sepet değişiminde yok olmuş bir component'in kodu çalışır. Component'i her açışta bir abonelik daha birikir. Angular'ın sonraki sürümleri bu deseni kısaltan araçlar getirdi; onları sürüm yazılarında göreceğiz.

Küçük Bir Deneme

Bir MouseTrackerComponent yaz. ngOnInit'te fromEvent<MouseEvent>(document, 'mousemove') akışına abone ol, map ile sadece clientX ve clientY değerlerini al, tap ile console'a yaz ve koordinatları ekranda göster. Component'i AppComponent'te bir butonla *ngIf üzerinden açıp kapat.

Önce takeUntil olmadan dene: component'i kapattıktan sonra fareyi oynatınca console'un yazmaya devam ettiğini göreceksin. Sonra destroy$ desenini ekle ve farkı izle. Son adımda manuel subscribe'ı tamamen kaldırıp koordinatları async pipe ile göstermeyi dene.

Bu yazıdaki bütün "sunucu" cevaplarını of ve delay ile taklit ettik. Gerçek bir API'ye istek atıp cevabı observable olarak almak, hataları yakalamak ve her isteğe token eklemek için Angular'ın kendi aracı var. Bir sonraki yazıda HttpClient ile sunucuyla konuşmaya başlıyoruz.