Angular 22'de RxJS: HttpClient, Router ve Forms Observable'ları, Signal Interop ve Subscription Yönetimi

Sürüm notu: Bu yazıdaki örnekler Angular 22.2, RxJS 7.8.2, TypeScript 6.0 ve Node.js 24 ile denenmiştir (Angular projesi, Angular 22.2'nin desteklediği TypeScript 6.0'ı kullanıyor); bütün Angular örnekleri Playwright 1.65 ile headless Chromium'da, snapshot'taki sahte Bookstore API'sine karşı çalıştırılmıştır.

Önce Tahmin Et

Sitedeki serilerin ortak örnek uygulaması Bookstore'da (bir online kitapçı) 1042 numaralı siparişin durumunu gösteren bir "My order" sayfası var. Polling'i zaman operator'leri yazısında yazmıştık; bookstore-web'deki OrderService.status$ artık aynısını HttpClient ile yapıyor (timer(0, 5000), exhaustMap, takeWhile). Sayfanın ilk sürümü constructor'da subscribe oluyor ve her cevabı bir field'a yazıyor:

@Component({
  selector: 'app-order-status-page',
  template: `
    <h1>Order 1042</h1>
    <p class="status">Status: {{ status?.status ?? 'Loading...' }}</p>
  `,
})
export class OrderStatusPage {
  protected status: OrderStatus | undefined;

  constructor() {
    inject(OrderService)
      .status$(1042)
      .subscribe((order) => {
        this.status = order;
        console.log(`[order] status ${order.status}`);
      });
  }
}

Sayfa 11 saniye açık kalıyor, sonra kullanıcı "Catalog"a tıklıyor. İki soru: 11 saniye sonra sayfada ne görünür ve kullanıcı ayrıldıktan sonra polling'e ne olur?

Bu yazı seriyi Angular 22'ye taşıyor. Hangi Angular API'lerinin Observable verdiği ve signal tabanlı karşılıklarının ne olduğu; toSignal, toObservable, rxResource, httpResource ve outputFromObservable; bir Observable'ın ekranı neden artık kendiliğinden güncellemediği; Bookstore'un üç şekilde yazılmış arama kutusu ve sızıntısız subscription yönetimi. Signal'ların ve takeUntilDestroyed'ın temeli Angular serisinin "Angular 16 Yenilikleri: Signals, Required Input ve takeUntilDestroyed" yazısında, Angular 22'nin varsayılanları "Angular 22: Varsayılan OnPush, Stable Signal Forms ve Angular 12 ile Proje Farkı" yazısında; burada RxJS'in bu tabloya nasıl oturduğuna bakıyoruz.

Zoneless ve OnPush: Observable Ekranı Kendiliğinden Güncellemez

Sayfayı headless Chromium'da açtık:

 0.1 s console: [analytics] page view /orders/1042
 0.1 s request /api/orders/1042/status
 0.3 s console: [order] status Paid
 5.1 s request /api/orders/1042/status
 5.3 s console: [order] status Paid
10.1 s request /api/orders/1042/status
10.3 s console: [order] status Paid
11.1 s screen: Status: Loading...
11.1 s console: [analytics] page view /
11.1 s left the order page
15.1 s request /api/orders/1042/status
15.3 s console: [order] status Shipped
20.1 s request /api/orders/1042/status
20.3 s console: [order] status Shipped

İki cevap da kötü. Üç cevap geldi ve konsol her birini yazdı, ama ekran hâlâ "Loading..." diyor. Kullanıcı ayrıldıktan sonra da polling devam etti: artık var olmayan bir sayfa için 15. ve 20. saniyelerde iki istek daha; sipariş teslim edilene kadar da sürecekti.

İlk sorun Angular 22'nin varsayılanlarından geliyor. Yeni bir proje zoneless, yani her asenkron callback'ten sonra change detection başlatmak için setTimeout'u ya da fetch'i patch'leyen bir şey yok; component'ler de aksi söylenmedikçe OnPush. Angular bir view'ı, bir şey ona söylediğinde yeniliyor: bir template event'i, template'te okunan ve değişen bir signal, async pipe ya da açık bir markForCheck. subscribe içinde düz bir field'a değer atamak bunların hiçbiri değil. Zone tabanlı eski bir uygulamada aynı kod tesadüfen çalışıyordu; bu kadar yaygın olmasının sebebi de bu. İkinci sorun Angular 22'den eski: elle yapılan bir subscribe, Observable tamamlanana kadar yaşıyor ve burada onu component'in ömrüne bağlayan hiçbir şey yok.

Düzeltilmiş sayfa tek satır:

import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { OrderService } from './order-service';

@Component({
  selector: 'app-order-status-page',
  template: `
    <h1>Order 1042</h1>
    <p class="status">Status: {{ status()?.status ?? 'Loading...' }}</p>
  `,
})
export class OrderStatusPage {
  // toSignal şimdi subscribe olur, sayfa destroy olunca unsubscribe olur
  protected readonly status = toSignal(inject(OrderService).status$(1042));
}

Template artık status()?.status okuyor. Aynı test:

 0.1 s console: [analytics] page view /orders/1042
 0.1 s request /api/orders/1042/status
 5.1 s request /api/orders/1042/status
10.1 s request /api/orders/1042/status
11.1 s screen: Status: Paid
11.1 s console: [analytics] page view /
11.1 s left the order page

Ekran "Paid" gösteriyor ve son istek 10. saniyedeki: kullanıcı ayrılınca toSignal unsubscribe oldu, exhaustMap timer'dan unsubscribe oldu ve polling durdu. Bu yazının genel kuralı da bu: zoneless, OnPush bir uygulamada bir Observable ekrana async pipe ya da toSignal üzerinden ulaşır ve ikisi de temizliği senin yerine yapar.

Angular API'lerindeki Observable'lar ve Signal Karşılıkları

RxJS, Angular'ın public API'sinin birkaç yerinde duruyor. Angular 16'dan beri çoğunun signal tabanlı bir alternatifi var; Angular 22 resource API'lerini ve Signal Forms'u stable olarak işaretliyor:

bookstore-web'den ikisi. Router.events'in signal sürümü yok ve buna ihtiyacı da yok: bir event akışı, tam olarak RxJS'in işi. Root component biten her navigation'ı, örneğin analytics için logluyor:

  constructor() {
    // Router.events bir Observable: biten her navigation için bir satır, ör. analytics için
    inject(Router)
      .events.pipe(
        filter((event) => event instanceof NavigationEnd),
        map((event) => event.urlAfterRedirects),
        takeUntilDestroyed(),
      )
      .subscribe((url) => console.log(`[analytics] page view ${url}`));
  }

TypeScript 5.5'ten beri derleyici event instanceof NavigationEnd'in tipi daralttığını kendisi çıkarıyor; map bu yüzden bir type guard olmadan urlAfterRedirects'i okuyabiliyor. Kitap detay sayfası ise artık ActivatedRoute'a ihtiyaç duymuyor: provideRouter(routes, withComponentInputBinding()) ile :isbn parametresi bir signal input olarak geliyor ve rxResource kitabı her değiştiğinde yüklüyor:

import { HttpClient } from '@angular/common/http';
import { Component, inject, input } from '@angular/core';
import { rxResource } from '@angular/core/rxjs-interop';
import { RouterLink } from '@angular/router';
import { apiUrl } from '../api';
import type { Book } from './book';

@Component({
  selector: 'app-book-page',
  imports: [RouterLink],
  template: `
    @if (book.value(); as b) {
      <h1>{{ b.title }}</h1>
      <p>{{ b.author }} - {{ b.price }} TRY - {{ b.stock }} in stock</p>
    } @else if (book.isLoading()) {
      <p>Loading...</p>
    }
    <a routerLink="/">Back to catalog</a>
  `,
})
export class BookPage {
  readonly #http = inject(HttpClient);
  // withComponentInputBinding() tarafından :isbn route parametresinden doldurulur
  readonly isbn = input.required<string>();

  protected readonly book = rxResource({
    params: () => this.isbn(),
    stream: ({ params: isbn }) => this.#http.get<Book>(`${apiUrl}/books/${isbn}`),
  });
}

console: [analytics] page view /
console: [analytics] page view /books/9780441172719
request /api/books/9780441172719
screen: Dune | Frank Herbert - 450 TRY - 12 in stock
console: [analytics] page view /
console: [analytics] page view /books/9780141439587
request /api/books/9780141439587
screen: Emma | Jane Austen - 320 TRY - 8 in stock

rxResource, reaktif bir fonksiyon olan params'ı ve params'ı bir Observable'a çeviren stream'i alıyor. isbn değişince resource önceki Observable'dan unsubscribe oluyor (switchMap gibi) ve yenisine subscribe oluyor; value(), isLoading(), error() ve status() template'in okuyabileceği signal'lar. Angular 22.2'nin type tanımları bir kombinasyona karşı da uyarıyor: Observable değer vermeden tamamlanırsa, örneğin catchError(() => EMPTY) yüzünden, resource NG0991 ile başarısız oluyor. Bunun yerine error handling yazısındaki gibi yedek bir değer döndür.

toSignal ve toObservable

toSignal hemen, bir injection context içinde subscribe oluyor ve o context (component, service) destroy olunca unsubscribe oluyor. Seçenekleri, signal'ın ilk değerden önce ne tuttuğunu belirliyor. Seçeneksiz tip T | undefined, yukarıdaki sipariş sayfasındaki gibi. initialValue bir başlangıç değeri veriyor, katalogdaki toSignal(…, { initialValue: [] }) gibi. requireSync: true, subscribe anında senkron yayan Observable'lar için (bir BehaviorSubject, startWith ile başlayan bir pipe) tipten undefined'ı çıkarıyor ve yaymazlarsa hata fırlatıyor. manualCleanup: true subscription'ı context destroy olduktan sonra da, Observable tamamlanana kadar tutuyor; bir component'in kendisinden uzun yaşaması gereken bir işi başlattığı nadir durumlar için. Error handling yazısında gördüğümüz gibi Observable'daki bir hata, signal her okunduğunda yeniden fırlatılıyor; hataları toSignal'dan önce, pipe içinde ele al.

toObservable ters yönde çalışıyor. Signal'ı bir effect içinde okuyor ve önce güncel değeri, sonra her değişikliği yayıyor. Aşağıdaki arama örneğinde bunun iki sonucu görünüyor: değerler set çağrıldığı anda değil asenkron geliyor ve aynı tur içindeki birkaç hızlı değişiklik tek bir yayına dönüşebiliyor; bu bir arama kutusu için sorun değil, bir event kanalı için yanlış. İlk değer, yani boş başlangıç metni de pipe'tan geçiyor; bizim koşumuzda searched output'u sayfa açılır açılmaz bir kez "" ile tetiklendi.

Arama Kutusu Üç Şekilde

bookstore-web'in /search sayfasında aynı arama kutusu üç kez var. Üçü de son tuştan sonra 300 ms bekliyor, tekrarlanan terimleri yok sayıyor ve sahte API'ye GET /api/books?q=… soruyor. Önce yalnız RxJS, bir reactive form control'den başlayarak:

import { AsyncPipe } from '@angular/common';
import { HttpClient } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { catchError, debounceTime, distinctUntilChanged, map, of, switchMap } from 'rxjs';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

// 1. Yalnız RxJS: form control'ün valueChanges'i, sonucu async pipe render ediyor
@Component({
  selector: 'app-search-rxjs',
  imports: [ReactiveFormsModule, AsyncPipe],
  template: `
    <input id="search-rxjs" [formControl]="term" placeholder="Search books" />
    <ul>
      @for (book of results$ | async; track book.isbn) {
        <li>{{ book.title }}</li>
      }
    </ul>
  `,
})
export class SearchRxjs {
  readonly #http = inject(HttpClient);
  protected readonly term = new FormControl('', { nonNullable: true });
  protected readonly results$ = this.term.valueChanges.pipe(
    debounceTime(300),
    map((term) => term.trim()),
    distinctUntilChanged(),
    switchMap((term) =>
      term
        ? this.#http
            .get<Book[]>(`${apiUrl}/books`, { params: { q: term } })
            .pipe(catchError(() => of<Book[]>([])))
        : of<Book[]>([]),
    ),
  );
}

İkincisinde state de sonuç da signal; yalnız zamanla ilgili kısım RxJS'ten geçiyor. outputFromObservable, oturmuş terimi normal bir component output'una çeviriyor:

import { HttpClient } from '@angular/common/http';
import { Component, inject, signal } from '@angular/core';
import { outputFromObservable, toObservable, toSignal } from '@angular/core/rxjs-interop';
import { catchError, debounceTime, distinctUntilChanged, map, of, switchMap } from 'rxjs';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

// 2. signal girer, signal çıkar: RxJS yalnız zamanla ilgili kısımda
@Component({
  selector: 'app-search-interop',
  template: `
    <input id="search-interop" #box [value]="term()" (input)="term.set(box.value)" />
    <ul>
      @for (book of results(); track book.isbn) {
        <li>{{ book.title }}</li>
      }
    </ul>
  `,
})
export class SearchInterop {
  readonly #http = inject(HttpClient);
  protected readonly term = signal('');

  readonly #settledTerm$ = toObservable(this.term).pipe(
    debounceTime(300),
    map((term) => term.trim()),
    distinctUntilChanged(),
  );

  protected readonly results = toSignal(
    this.#settledTerm$.pipe(
      switchMap((term) =>
        term
          ? this.#http
              .get<Book[]>(`${apiUrl}/books`, { params: { q: term } })
              .pipe(catchError(() => of<Book[]>([])))
          : of<Book[]>([]),
      ),
    ),
    { initialValue: [] },
  );

  // parent, her output() gibi (searched)="…" ile dinleyebilir
  readonly searched = outputFromObservable(this.#settledTerm$);
}

Üçüncüsünde hiç RxJS yok: Angular 22'nin type tanımlarında 22.0'dan beri experimental olarak işaretlenen debounced(), signal'ı 300 ms geriden takip eden bir resource üretiyor; httpResource isteği onun değerinden kuruyor. URL olarak undefined döndürmek "istek yok" demek:

import { httpResource } from '@angular/common/http';
import { Component, debounced, signal } from '@angular/core';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

// 3. RxJS yok: debounced() (Angular 22'de experimental) ve httpResource
@Component({
  selector: 'app-search-resource',
  template: `
    <input id="search-resource" #box [value]="term()" (input)="term.set(box.value)" />
    @if (results.isLoading()) {
      <p>Searching...</p>
    }
    <ul>
      @for (book of results.value(); track book.isbn) {
        <li>{{ book.title }}</li>
      }
    </ul>
  `,
})
export class SearchResource {
  protected readonly term = signal('');
  readonly #settledTerm = debounced(this.term, 300);

  protected readonly results = httpResource<Book[]>(
    () => {
      const term = this.#settledTerm.value().trim();
      return term ? `${apiUrl}/books?q=${encodeURIComponent(term)}` : undefined;
    },
    { defaultValue: [] },
  );
}

Playwright her kutuya 100 ms arayla "dune" yazdı, 350 ms sonra bir "s" ve hemen ardından bir backspace ekledi; yani terim değişip "dune"a geri döndü:

  396 ms console: [search] searched for ""
--- page loaded, nothing typed yet
--- 1. #search-rxjs: type "dune", one key every 100 ms, then "s" 350 ms later and backspace
  641 ms request /api/books?q=dune
 1703 ms screen: ["Dune"] 
--- 2. #search-interop: type "dune", one key every 100 ms, then "s" 350 ms later and backspace
  632 ms request /api/books?q=dune
  633 ms console: [search] searched for "dune"
 1688 ms screen: ["Dune"] 
--- 3. #search-resource: type "dune", one key every 100 ms, then "s" 350 ms later and backspace
  641 ms request /api/books?q=dune
 1699 ms screen: ["Dune"] | "Searching..." seen while typing: true

Üçü de son tuştan yaklaşık 300 ms sonra "dune" için tam bir istek gönderdi; "dunes" 100 ms yaşadı ve API'ye hiç ulaşmadı. İlk ikisi backspace'ten sonra "dune"u tekrar aramadı, çünkü distinctUntilChanged var; httpResource aramadı, çünkü hesapladığı URL aynıydı. Üçüncü kutunun ilk sürümü debounce'lu terimi .value() yerine params context'indeki chain(this.#settledTerm) ile okuyordu. chain, debounce'un bekleme durumunu HTTP resource'una aktarıyor; bu yüzden kullanıcı yazarken "Searching..." gösterdi ve backspace'ten sonra "dune"u ikinci kez istedi. İkisi de istenen davranış olabilir, ama bilerek seçilmeli.

Tablonun arkasındaki ayrım, bu serinin Angular 22 için kuralı: zaman ve event akışı (debounce, iptal, birleştirme, tekrar deneme) RxJS'e ait; ekranın gösterdiği state signal'lara. Dönüşüm kenarlarda, bir kez, toSignal ya da toObservable ile yapılıyor.

Subscription Yönetimi

Modern bir Angular component'indeki subscription'ların çoğu kodda hiç görünmüyor: async pipe, toSignal, rxResource ve httpResource senin yerine subscribe ve unsubscribe oluyor. Geriye kalan elle yazılmış subscribe çağrıları için takeUntilDestroyed(), subscription'ı component destroy olunca bitiriyor. Bunu, o anki injection context'in DestroyRef'ini arayarak yapıyor; bu da yalnız constructor'da, field initializer'da ya da bir factory'de var. Bir test component'inde onu ngOnInit içinde, toSignal'ı da bir template method'unda kullandık:

broken-a [console.error] ERROR RuntimeError: NG0602: toSignal() cannot be called from within a reactive context. Invoking `toSignal` causes new subscriptions every time. Consider moving `toSignal` outside of the reactive context and read the signal value where needed. Find more at https://v22.angular.dev/errors/NG0602
broken-b [console.error] ERROR RuntimeError: NG0203: takeUntilDestroyed() can only be used within an injection context such as a constructor, a factory function, a field initializer, or a function used with `runInInjectionContext`. Find more at https://v22.angular.dev/errors/NG0203

Injection context dışında DestroyRef'i kendin ver; bu sürüm aynı test component'inde hatasız çalıştı:

readonly #destroyRef = inject(DestroyRef);

ngOnInit(): void {
  interval(1000)
    .pipe(takeUntilDestroyed(this.#destroyRef))
    .subscribe((n) => this.ticks.set(n));
}

NG0602 hatası yaygın bir yanlışı yakalıyor: template'in her change detection'da çalıştırdığı bir method'da çağrılan toSignal her seferinde yeni bir subscription oluştururdu. Angular 22 buna izin vermiyor; signal'ı bir kez, bir field'da oluştur.

Code review'da gördüğümüz sızıntıların çoğunu iki pattern daha üretiyor. İlki root service'teki bitmeyen bir kaynak: bir @Service() içinde başlatılan bir interval ya da WebSocket uygulama kadar yaşıyor ve temizlik yapmadan subscribe olan her component ona kendi subscription'ını tutuyor. Böyle bir kaynak, multicasting yazısındaki shareReplay({ bufferSize: 1, refCount: true }) ile paylaşılmalı ki dinleyen kalmayınca dursun. İkincisi iç içe subscribe:

// anti-pattern: subscribe içinde subscribe
this.#route.paramMap.subscribe((params) => {
  this.#http.get<Book>(`${apiUrl}/books/${params.get('isbn')}`).subscribe((book) => this.book.set(book));
});

Her şey yolunda giderken çalışıyor, ama dış subscription hiç temizlenmiyor, önceki kitap için gelen yavaş bir istek güncel olanın üzerine yazabiliyor ve içteki istekler iptal edilemiyor. Bu, garantileri olmadan elle yazılmış bir switchMap; yukarıdaki kitap sayfası aynı işi rxResource ile yapıyor, saf RxJS sürümü ise paramMap.pipe(switchMap(…)) olurdu.

Üretimde Dikkat

Angular DevTools signal'ları ve bağımlılıklarını gösteriyor, ama RxJS subscription'larını göstermiyor. Sızıntı bulmak için sayfalar arasında birkaç kez gidip gel ve Network sekmesinde, tahmin örneğindeki gibi, ayrıldığın sayfalara ait istekleri ara; ya da Chrome DevTools'ta heap snapshot'lar alıp sayısı sürekli artan component instance'larına bak. Geliştirme sırasında finalize içinde loglamak (ya da custom operator yazısının konusu olan bir debug operator'ü) her subscription'ın tam olarak ne zaman bittiğini gösteriyor.

Var olan bir uygulamayı zoneless'a taşımak, tahmin örneğindeki eskimiş ekran sorununu çoğu zaman birçok yerde birden ortaya çıkarıyor. Component'lerde subscribe( araması yap ve her birini kontrol et: template'e giden bir değer toSignal, async pipe ya da bir signal.set üzerinden gitmeli. Angular'ın provideCheckNoChangesConfig({ exhaustive: true })'ı geliştirme sırasında Angular'a haber verilmeden değişen binding'leri bildirerek yardım ediyor.

Son olarak production kodunda experimental API'lerle dikkatli ol. debounced() bir minor sürümde hâlâ değişebilir; kullanıyorsan küçük bir wrapper'ın arkasında tut, böylece toObservable + debounceTime'a geçmek tek bir dosyaya dokunur.

Sık Düşülen Tuzaklar

• Zoneless, OnPush bir component'te bir Observable'ın değerini subscribe içinde düz bir field'a atamak. Ekran güncellenmez; toSignal, async ya da bir signal kullan.
• takeUntilDestroyed olmadan elle subscribe. Polling ve timer'lar kullanıcı sayfadan çıktıktan sonra da sürer.
• takeUntilDestroyed()'ı ngOnInit'te ya da bir event handler'da argümansız çağırmak. Injection context dışında NG0203 fırlatır; bir DestroyRef ver.
• toSignal'ı template'in kullandığı bir method ya da getter içinde çağırmak. Angular 22 NG0602 fırlatır; signal'ı bir kez, bir field'da oluştur.
• Observable'ı başarısız olabilen bir toSignal signal'ını catchError olmadan okumak. Her okuma hatayı yeniden fırlatır.
• subscribe çağrılarını iç içe yazmak. switchMap, rxResource ya da httpResource kullan.
• toObservable'ı bir event kanalı olarak kullanmak. Değişiklikler asenkron gelir ve birleşebilir; event'ler bir Subject'e ya da bir output()'a ait.

Küçük Bir Deneme

★☆☆ search-interop.ts'te outputFromObservable'dan önce filter((term) => term !== '') ekle, böylece searched output'u boş başlangıç metni için tetiklenmesin. Kullanıcı terimin tamamını silince sonuç listesi hâlâ temizleniyor mu? Temizlenmiyorsa filter nereye konmalı?

★★☆ Kitap sayfasını withComponentInputBinding ve rxResource olmadan, ActivatedRoute.paramMap, switchMap ve toSignal ile yeniden yaz. Sahte API'yi DELAY_MS=1500 ile çalıştırıp Dune'dan Emma'ya hızlıca geç ve Network sekmesinde Dune isteğinin iptal edildiğini kontrol et.

★★★ "My order" sayfasına, hemen yeni bir durum isteği başlatan ve beş saniyelik timer'ı yeniden başlatan, ama hiçbir zaman aynı anda iki istek olmayan bir "Refresh now" butonu ekle. Bir kez RxJS ile (timer ve tıklama akışının merge'ünü exhaustMap'e vererek), bir kez de rxResource ve onun reload()'u ile yaz; buton bir istek sürerken tıklandığında iki sürümün ne yaptığını karşılaştır.

Bu serideki bazı pipe'lar kendini tekrar etmeye başladı: debounce'lu arama bloğu, backoff'lu retry, bir isteğin etrafındaki loading durumu. Bir sonraki yazıda bu blokları pipe() ve OperatorFunction ile custom operator'lere çevirecek, new Observable ile sıfırdan bir tane yazacak ve uzun pipe'ları isimlendirme, bir debug operator'ü ve RxJS için ESLint kurallarıyla okunur tutacağız.