Zamanla Çalışan Operator'ler ve Scheduler'lar: debounceTime, throttleTime, auditTime, timeout ve Polling

Sürüm notu: Bu yazıdaki örnekler RxJS 7.8.2, TypeScript 6.0, Node.js 24 ve Vite 8.3 ile denenmiştir; zamanlı örnekler RxJS'in TestScheduler'ı ile sanal zamanda, tarayıcı örnekleri Playwright 1.65 ile headless Chromium'da çalıştırılmıştır.

Önce Tahmin Et

Sitedeki serilerin ortak örnek uygulaması Bookstore'da (bir online kitapçı) müşteriler kitaplara yorum yazabiliyor. Ada, Dune için bir yorum yazıyor; taslak otomatik kaydedilmeli, ama her tuş vuruşunda değil. Kaydetmenin önüne debounceTime(500) koyuyoruz:

import { concat, debounceTime, NEVER } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";

runInVirtualTime("review draft autosave", () => {
  // Ada, Dune için yorum yazıyor: üç hızlı düzenleme, bir duraklama, iki düzenleme daha
  const draft$ = concat(
    emitAt(
      [0, "Great"],
      [200, "Great book"],
      [400, "Great book, slow start"],
      [1500, "Great book, slow start but"],
      [1700, "Great book, slow start but worth it"],
    ),
    NEVER, // metin kutusu açık kalıyor: akış hiç tamamlanmıyor
  );

  draft$.pipe(debounceTime(500)).subscribe((text) => log(`save: "${text}"`));
});

Kaç kayıt görürsün, hangi milisaniyelerde ve hangi metinle? "Great" hiç kaydedilir mi?

Önceki yazıda zamanlamayı dışarısı belirledi: kullanıcı ne zaman tıkladı, API ne kadar sürede cevap verdi. Bu yazıda zamanı kendimiz yönetiyoruz. debounceTime, throttleTime, auditTime, sampleTime, delay ve bufferTime'ı aynı girdiyle karşılaştıracak, timeout'u doğru yere koyacak, sayfa gizliyken duran bir polling yazacak ve RxJS'in scheduler'larının event loop'ta gerçekte ne yaptığına bakacağız. Zamanlı örnekler önceki yazılardaki runInVirtualTime, emitAt ve log yardımcılarıyla sanal zamanda çalışıyor, yani her koşuda aynı milisaniyeleri basıyor; tarayıcı örnekleri snapshot'taki sahte Bookstore API'sine bağlanıyor.

Aynı Girdiyle Altı Zaman Operator'ü

Farkı görmenin en hızlı yolu, bütün operator'lere aynı marble'ı vermek. Girdi hızlı bir seri (A, B, C), bir duraklama ve ardından her 3 ms'de bir gelen değerlerden oluşuyor; hızlı yazıp duran, sonra düzenli yazmaya devam eden bir kullanıcı gibi:

import { asyncScheduler, auditTime, bufferTime, debounceTime, delay, sampleTime, throttleTime } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

// bir karakter = 1 ms sanal zaman: hızlı bir seri (A B C), duraklama, sonra her 3 ms'de bir değer
const keys = "ABC------D--E--F--G---|";

showMarbles("debounceTime(4)", (cold) => cold(keys).pipe(debounceTime(4)), 24);
showMarbles("throttleTime(4)", (cold) => cold(keys).pipe(throttleTime(4)), 24);
showMarbles("throttleTime(4) trailing", (cold) => cold(keys).pipe(throttleTime(4, asyncScheduler, { leading: false, trailing: true })), 24);
showMarbles("auditTime(4)", (cold) => cold(keys).pipe(auditTime(4)), 24);
showMarbles("sampleTime(5)", (cold) => cold(keys).pipe(sampleTime(5)), 24);
showMarbles("delay(2)", (cold) => cold(keys).pipe(delay(2)), 24);
showMarbles("bufferTime(5)", (cold) => cold(keys).pipe(bufferTime(5)), 24);

debounceTime(4)
  in : ABC------D--E--F--G---| 
  out: ------C---------------(G|) 
throttleTime(4)
  in : ABC------D--E--F--G---| 
  out: A--------D-----F------| 
throttleTime(4) trailing
  in : ABC------D--E--F--G---| 
  out: ----C--------E---F---G| 
auditTime(4)
  in : ABC------D--E--F--G---| 
  out: ----C--------E-----G--| 
sampleTime(5)
  in : ABC------D--E--F--G---| 
  out: -----C----D----F----G-| 
delay(2)
  in : ABC------D--E--F--G---| 
  out: --ABC------D--E--F--G-| 
bufferTime(5)
  in : ABC------D--E--F--G---| 
  out: -----a----b----c----d-(e|) a=[ 'A', 'B', 'C' ] b=[ 'D' ] c=[ 'E', 'F' ] d=[ 'G' ] e=[]

Çıktıyı satır satır oku, çünkü her operator farklı bir soruya cevap veriyor:

Marble'larda üç ayrıntı önemli. debounceTime ikinci seri boyunca hiçbir şey yaymadı: değerler 3 ms'de bir geldi, pencere 4 ms'ydi, yani kaynak hiç yeterince uzun sessiz kalmadı. G'yi ancak 22. frame'deki complete serbest bıraktı, | ile birlikte. Düzenli yazan bir kullanıcı debounce'lu bir aramayı hiç tetiklemez; istenen davranış bu, ama hiç tamamlanmayan bir debounce'lu akışın son değerini sonsuza kadar tutabilmesinin sebebi de bu. trailing'li throttleTime ile auditTime birbirine benziyor ama aynı değil: 13. frame'de E'yi yaydıktan sonra throttle hemen yeni bir pencere açtı ve 17'de F'yi yaydı; audit ise pencere açmak için bir sonraki değeri (15'teki F) bekledi ve 19'da en sonuncuyu, G'yi yaydı. bufferTime ise sonda hiçbir şey gelmediği hâlde boş bir array yaydı; her buffer'ı sunucuya gönderiyorsan boş olanları süz.

debounceTime ile Otomatik Kaydetme ve Kaybolan Düzenleme

Baştaki sorunun cevabı:

--- review draft autosave
  900 ms  save: "Great book, slow start"
 2200 ms  save: "Great book, slow start but worth it"

İki kayıt. "Great" ve "Great book" sunucuya hiç gitmedi, çünkü her seferinde 500 ms dolmadan yeni bir düzenleme geldi; 900. ms'de metin 500 ms boyunca değişmemişti ve son hâli kaydedildi. Aynısı ikinci seriden sonra 2200. ms'de oldu. Bir taslağın ihtiyacı tam olarak bu: daha az istek ve her zaman son hâlin kazanması.

Gerçek bir otomatik kaydetmenin iki ihtiyacı daha var. Ada bir harf yazıp geri silerse metin son kayıtla aynıdır, distinctUntilChanged bunu atlamalı. Kaydetmenin kendisi de bir yazma işlemi: higher-order mapping yazısında gördüğümüz gibi switchMap yoldaki bir PUT'u iptal edebilir, bu yüzden kayıtlar concatMap ile sırayla gidiyor. Asıl ilginç soru, Ada son düzenlemesinden 300 ms sonra, o düzenleme hâlâ debounceTime'ın içinde beklerken sayfadan çıkarsa ne olduğu:

import { concat, concatMap, debounceTime, distinctUntilChanged, NEVER, takeUntil } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

function saveDraft(text: string) {
  return fakeRequest(`PUT /reviews/dune/draft (${text.length} chars)`, 300, () => text);
}

// Ada'nın yorum kutusuna yazdıkları (kutu açık kalıyor, akış hiç tamamlanmıyor)
function typing() {
  return concat(
    emitAt(
      [0, "Great book"],
      [300, "Great book, slow start"],
      [1400, "Great book, slow start"], // yazılıp geri silinen bir harf: metin aynı
      [2600, "Great book, slow start but worth it"],
    ),
    NEVER,
  );
}

const report = { next: (text: string) => log(`saved: "${text}"`), complete: () => log("autosave stopped") };

runInVirtualTime("takeUntil after debounceTime", () => {
  const leave$ = emitAt([2900, "leave"]); // Ada son düzenlemesinden 300 ms sonra "Back to catalog"a tıklıyor
  typing()
    .pipe(
      debounceTime(800),
      distinctUntilChanged(),
      concatMap((text) => saveDraft(text)),
      takeUntil(leave$), // sayfadan çıkmak, debounceTime'da bekleyen düzenlemeyi atar
    )
    .subscribe(report);
});

runInVirtualTime("takeUntil before debounceTime", () => {
  const leave$ = emitAt([2900, "leave"]);
  typing()
    .pipe(
      takeUntil(leave$), // kaynak tamamlanıyor, debounceTime bekleyen düzenlemeyi hemen yayıyor
      debounceTime(800),
      distinctUntilChanged(),
      concatMap((text) => saveDraft(text)),
    )
    .subscribe(report);
});

--- takeUntil after debounceTime
 1100 ms  -> PUT /reviews/dune/draft (22 chars)
 1400 ms  <- PUT /reviews/dune/draft (22 chars)
 1400 ms  saved: "Great book, slow start"
 2900 ms  autosave stopped
--- takeUntil before debounceTime
 1100 ms  -> PUT /reviews/dune/draft (22 chars)
 1400 ms  <- PUT /reviews/dune/draft (22 chars)
 1400 ms  saved: "Great book, slow start"
 2900 ms  -> PUT /reviews/dune/draft (35 chars)
 3200 ms  <- PUT /reviews/dune/draft (35 chars)
 3200 ms  saved: "Great book, slow start but worth it"
 3200 ms  autosave stopped

İlk sürümde son cümle, "but worth it", kayboldu: sondaki takeUntil her şeyden unsubscribe oldu, debounceTime'ın elinde tuttuğu değer de dahil. İkinci sürümde aynı operator debounceTime'ın önünde duruyor. Bu kez sayfadan çıkmak kaynağı tamamlıyor ve debounceTime, kaynak tamamlanır tamamlanmaz bekleyen değeri 800 ms'nin kalanını beklemeden aşağı gönderiyor. Kayıt 2900. ms'de gidiyor, akış da sunucu cevap verdikten sonra tamamlanıyor.

Arkasındaki kural akılda kalıcı: unsubscribe bekleyen değeri atar, complete onu gönderir. Angular component'inde bunun önemi şu: component destroy olunca unsubscribe gerçekleşir; son taslağın mutlaka kaydedilmesi gerekiyorsa ya yukarıdaki gibi kaynağı bir çıkış event'iyle tamamla ya da kaydetmeyi component'ten uzun yaşayan bir service'e taşı.

timeout: first, each, with ve Yanlış Yer

RxJS 7'nin timeout'u bir config object alıyor. first ilk değer için bekleme süresini, each herhangi iki değer arasındaki boşluğu sınırlıyor; with ise zaman aşımında ne olacağını belirliyor: verilmezse TimeoutError gelir, verilirse akış senin döndürdüğün Observable'a geçer.

import { concat, debounceTime, NEVER, of, switchMap, timeout } from "rxjs";
import { books } from "../../data/books.ts";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

const dune = books[0];

runInVirtualTime("timeout first + with: fall back to the cached stock", () => {
  const cachedStock = { isbn: dune.isbn, stock: 12, cached: true };
  fakeRequest(`GET /books/${dune.isbn}/stock`, 3000, () => ({ isbn: dune.isbn, stock: 11, cached: false }))
    .pipe(timeout({ first: 1000, with: () => of(cachedStock) }))
    .subscribe((s) => log(`stock ${s.stock} (cached: ${s.cached})`));
});

runInVirtualTime("timeout each: a live stream that goes silent", () => {
  // stok güncellemeleri 2 sn'de bir geliyor, depo akışı 6. sn'den sonra susuyor
  const stock$ = concat(emitAt([2000, 12], [4000, 11], [6000, 10]), NEVER);
  stock$.pipe(timeout({ each: 5000 })).subscribe({
    next: (n) => log(`stock ${n}`),
    error: (err: Error) => log(`${err.name}: ${err.message}`),
  });
});

function searchBooks(term: string) {
  return fakeRequest(`GET /books?q=${term}`, 300, () => books.filter((b) => b.title.toLowerCase().includes(term)).length);
}

runInVirtualTime("timeout outside switchMap: measures the user, not the request", () => {
  const terms$ = concat(emitAt([0, "dun"], [400, "dune"]), NEVER); // kullanıcı yazmayı bırakıyor, kutu açık kalıyor
  terms$
    .pipe(
      debounceTime(300),
      switchMap((term) => searchBooks(term)),
      timeout(2000), // yanlış yer
    )
    .subscribe({ next: (n) => log(`${n} result(s)`), error: (err: Error) => log(`search box dead: ${err.name}`) });
});

runInVirtualTime("timeout inside switchMap: measures each request", () => {
  const terms$ = concat(emitAt([0, "dun"], [400, "dune"]), NEVER);
  terms$
    .pipe(
      debounceTime(300),
      switchMap((term) => searchBooks(term).pipe(timeout(2000))),
    )
    .subscribe({ next: (n) => log(`${n} result(s)`), error: (err: Error) => log(`search box dead: ${err.name}`) });
});

--- timeout first + with: fall back to the cached stock
    0 ms  -> GET /books/9780441172719/stock
 1000 ms  x  GET /books/9780441172719/stock cancelled
 1000 ms  stock 12 (cached: true)
--- timeout each: a live stream that goes silent
 2000 ms  stock 12
 4000 ms  stock 11
 6000 ms  stock 10
11000 ms  TimeoutError: Timeout has occurred
--- timeout outside switchMap: measures the user, not the request
  300 ms  -> GET /books?q=dun
  600 ms  <- GET /books?q=dun
  600 ms  1 result(s)
  700 ms  -> GET /books?q=dune
 1000 ms  <- GET /books?q=dune
 1000 ms  1 result(s)
 3000 ms  search box dead: TimeoutError
--- timeout inside switchMap: measures each request
  300 ms  -> GET /books?q=dun
  600 ms  <- GET /books?q=dun
  600 ms  1 result(s)
  700 ms  -> GET /books?q=dune
 1000 ms  <- GET /books?q=dune
 1000 ms  1 result(s)

İlk senaryo kitap detay sayfasındaki stok sorgusu. Depo servisi bugün yavaş (3 saniye); bir saniye sonra isteği iptal edip son başarılı cevaptaki stoku gösteriyoruz ve ekranın "güncel olmayabilir" diyebilmesi için cache'ten geldiğini işaretliyoruz. x satırı, timeout'un istekten gerçekten unsubscribe olduğunu gösteriyor; HttpClient'ta bu, tarayıcının isteği iptal etmesi demek. İkinci senaryo canlı bir akışta each kullanıyor: depo akışı 6000. ms'deki son güncellemeden sonra beş saniye susunca akış 11000. ms'de hataya düşüyor ve hata tarafı yeniden bağlanabiliyor ya da bir uyarı gösterebiliyor.

Son iki senaryo timeout ile en sık yapılan hatayı gösteriyor. İkisi de aynı arama kutusunu kullanıyor ve istekler 300 ms sürüyor. İlkinde timeout(2000), switchMap'in dışında, dış akışta duruyor. İsteği ölçmüyor; iki arama sonucu arasındaki süreyi ölçüyor. Kullanıcı "dune"dan sonra yazmayı bıraktı, son sonuçtan iki saniye sonra dış akış zaman aşımına uğradı ve arama kutusu öldü: bundan sonra yazmak hiçbir şey yapmıyor. Aynı timeout switchMap'in içine taşınınca her isteği ayrı ölçüyor ve kullanıcı ne kadar beklerse beklesin kutu çalışmaya devam ediyor. timeout'u ölçmek istediğin şeyin yanına koy; HTTP için bu, içteki Observable.

Polling: timer, switchMap ve Üst Üste Binen İstek

Siparişlerim sayfasında 1042 numaralı siparişin durumu kendiliğinden güncellenmeli: Paid, Shipped, Delivered. Bilinen çözüm timer(0, 5000) ile bir flatten operator'ü. Hangisini seçtiğin, bir isteğin polling periyodundan uzun sürdüğünde ne olacağını belirliyor. Aşağıdaki örnekte ikinci çağrı 6,5 saniye, diğerleri 300 ms sürüyor; snapshot'taki statusApi (src/06/status-api.ts) sırasıyla Paid, Paid, Shipped, Shipped, Delivered cevabı veriyor:

import { defer, exhaustMap, mergeMap, repeat, switchMap, takeWhile, timer, type Observable } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { statusApi, type Status } from "./status-api.ts";

function poll(title: string, build: (getStatus: () => Observable<Status>) => Observable<Status>) {
  runInVirtualTime(title, () => {
    // ikinci çağrı yavaş (6,5 sn), diğerleri 300 ms
    build(statusApi([300, 6500]))
      .pipe(takeWhile((status) => status !== "Delivered", true)) // Delivered'dan sonra dur
      .subscribe((status) => log(`screen: ${status}`));
  });
}

poll("timer + mergeMap", (getStatus) => timer(0, 5000).pipe(mergeMap(() => getStatus())));
poll("timer + switchMap", (getStatus) => timer(0, 5000).pipe(switchMap(() => getStatus())));
poll("timer + exhaustMap", (getStatus) => timer(0, 5000).pipe(exhaustMap(() => getStatus())));
poll("defer + repeat({ delay })", (getStatus) => defer(() => getStatus()).pipe(repeat({ delay: 5000 })));

--- timer + mergeMap
    0 ms  -> GET /orders/1042/status #1
  300 ms  <- GET /orders/1042/status #1
  300 ms  screen: Paid
 5000 ms  -> GET /orders/1042/status #2
10000 ms  -> GET /orders/1042/status #3
10300 ms  <- GET /orders/1042/status #3
10300 ms  screen: Shipped
11500 ms  <- GET /orders/1042/status #2
11500 ms  screen: Paid
15000 ms  -> GET /orders/1042/status #4
15300 ms  <- GET /orders/1042/status #4
15300 ms  screen: Shipped
20000 ms  -> GET /orders/1042/status #5
20300 ms  <- GET /orders/1042/status #5
20300 ms  screen: Delivered
--- timer + switchMap
    0 ms  -> GET /orders/1042/status #1
  300 ms  <- GET /orders/1042/status #1
  300 ms  screen: Paid
 5000 ms  -> GET /orders/1042/status #2
10000 ms  x  GET /orders/1042/status #2 cancelled
10000 ms  -> GET /orders/1042/status #3
10300 ms  <- GET /orders/1042/status #3
10300 ms  screen: Shipped
15000 ms  -> GET /orders/1042/status #4
15300 ms  <- GET /orders/1042/status #4
15300 ms  screen: Shipped
20000 ms  -> GET /orders/1042/status #5
20300 ms  <- GET /orders/1042/status #5
20300 ms  screen: Delivered
--- timer + exhaustMap
    0 ms  -> GET /orders/1042/status #1
  300 ms  <- GET /orders/1042/status #1
  300 ms  screen: Paid
 5000 ms  -> GET /orders/1042/status #2
11500 ms  <- GET /orders/1042/status #2
11500 ms  screen: Paid
15000 ms  -> GET /orders/1042/status #3
15300 ms  <- GET /orders/1042/status #3
15300 ms  screen: Shipped
20000 ms  -> GET /orders/1042/status #4
20300 ms  <- GET /orders/1042/status #4
20300 ms  screen: Shipped
25000 ms  -> GET /orders/1042/status #5
25300 ms  <- GET /orders/1042/status #5
25300 ms  screen: Delivered
--- defer + repeat({ delay })
    0 ms  -> GET /orders/1042/status #1
  300 ms  <- GET /orders/1042/status #1
  300 ms  screen: Paid
 5300 ms  -> GET /orders/1042/status #2
11800 ms  <- GET /orders/1042/status #2
11800 ms  screen: Paid
16800 ms  -> GET /orders/1042/status #3
17100 ms  <- GET /orders/1042/status #3
17100 ms  screen: Shipped
22100 ms  -> GET /orders/1042/status #4
22400 ms  <- GET /orders/1042/status #4
22400 ms  screen: Shipped
27400 ms  -> GET /orders/1042/status #5
27700 ms  <- GET /orders/1042/status #5
27700 ms  screen: Delivered

mergeMap'te yavaş ikinci istek üçüncüsü başladığında hâlâ sürüyordu ve geç gelen cevabı üçüncünün arkasından geldi: 10300. ms'de ekran Shipped dedi, 11500. ms'de Paid'e geri döndü. Müşteri siparişin geriye gittiğini görüyor. Bu en sık görülen polling hatası ve hızlı bir yerel API'de ortaya çıkmıyor; yavaş bir mobil bağlantıda çıkıyor.

switchMap, bir sonraki tik geldiğinde yavaş isteği iptal etti, böylece ekrandaki sıra doğru kaldı. Bedeli şu: sunucu her zaman periyottan yavaşsa her istek iptal edilir ve ekran hiç güncellenmez. exhaustMap 10000. ms'deki tiki yok saydı, çünkü bir istek hâlâ sürüyordu; yani aynı anda hiçbir zaman birden fazla istek olmuyor. defer artı repeat({ delay: 5000 }) bir adım daha ileri gidiyor: her başlangıçtan değil, her cevaptan sonra beş saniye bekliyor; sunucu ne kadar yavaş olursa olsun istekler arasındaki boşluk aynı kalıyor. Durum polling'i için daha güvenli varsayılan exhaustMap ya da repeat({ delay }). takeWhile(…, true) son durumdan sonra polling'i durduruyor ve true sayesinde "Delivered"ı da ekrana geçiriyor.

Sayfa Gizliyken Polling'i Durdurmak

Sipariş sayfasını bir sekmede açıp unutan bir müşteri, bütün gün beş saniyede bir istek göndermemeli. Tarayıcı, sekme gizlendiğinde ya da tekrar göründüğünde document üzerinde visibilitychange event'i yayıyor; hangisi olduğunu document.visibilityState söylüyor. Bu bir boolean akışına dönüşüyor: switchMap, true için polling'i başlatıyor, false için EMPTY döndürüyor ve bu da çalışan timer'dan unsubscribe oluyor. Önce sanal zamanda:

import { distinctUntilChanged, EMPTY, exhaustMap, switchMap, takeWhile, tap, timer } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { statusApi } from "./status-api.ts";

runInVirtualTime("poll only while the page is visible", () => {
  const getStatus = statusApi();
  // tarayıcıda: fromEvent(document, "visibilitychange") → document.visibilityState === "visible"
  const visible$ = emitAt([0, true], [7000, false], [19000, true]);

  visible$
    .pipe(
      distinctUntilChanged(),
      tap((visible) => log(visible ? "page visible: polling starts" : "page hidden: polling stops")),
      switchMap((visible) => (visible ? timer(0, 5000).pipe(exhaustMap(() => getStatus())) : EMPTY)),
      takeWhile((status) => status !== "Delivered", true),
    )
    .subscribe((status) => log(`screen: ${status}`));
});

--- poll only while the page is visible
    0 ms  page visible: polling starts
    0 ms  -> GET /orders/1042/status #1
  300 ms  <- GET /orders/1042/status #1
  300 ms  screen: Paid
 5000 ms  -> GET /orders/1042/status #2
 5300 ms  <- GET /orders/1042/status #2
 5300 ms  screen: Paid
 7000 ms  page hidden: polling stops
19000 ms  page visible: polling starts
19000 ms  -> GET /orders/1042/status #3
19300 ms  <- GET /orders/1042/status #3
19300 ms  screen: Shipped
24000 ms  -> GET /orders/1042/status #4
24300 ms  <- GET /orders/1042/status #4
24300 ms  screen: Shipped
29000 ms  -> GET /orders/1042/status #5
29300 ms  <- GET /orders/1042/status #5
29300 ms  screen: Delivered

7. ve 19. saniyeler arasında hiç istek gitmedi. Sayfa tekrar görünür olunca polling hemen bir istekle baştan başladı; müşteri beş saniye beklemek yerine güncel durumu anında görüyor.

Aynı akış tarayıcıda, sahte API'ye karşı (snapshot'ta npm run web ve node api/server.ts; sayfa web/order-status.html). Sahte API, 1042 numaralı siparişin durum endpoint'ine üç kez Paid, sonra iki kez Shipped, sonra Delivered cevabı veriyor:

import { distinctUntilChanged, EMPTY, exhaustMap, fromEvent, map, startWith, switchMap, takeWhile, timer } from "rxjs";
import { fromFetch } from "rxjs/fetch";

type OrderStatus = { id: number; status: "Paid" | "Shipped" | "Delivered" };

const statusText = document.querySelector<HTMLParagraphElement>("#status")!;

const getStatus = () =>
  fromFetch("http://localhost:5180/api/orders/1042/status", {
    selector: (response) => response.json() as Promise<OrderStatus>,
  });

const visible$ = fromEvent(document, "visibilitychange").pipe(
  map(() => document.visibilityState === "visible"),
  startWith(document.visibilityState === "visible"),
  distinctUntilChanged(),
);

visible$
  .pipe(
    switchMap((visible) => (visible ? timer(0, 5000).pipe(exhaustMap(() => getStatus())) : EMPTY)),
    takeWhile((order) => order.status !== "Delivered", true),
  )
  .subscribe((order) => {
    statusText.textContent = `Order ${order.id}: ${order.status}`;
    console.log(statusText.textContent);
  });

Sayfayı Playwright ile headless Chromium'da çalıştırdık. Headless Chromium bir sayfayı kendiliğinden hiç gizlemiyor; bu yüzden test script'i 7. ve 19. saniyede document.visibilityState'i değiştirip gerçek bir visibilitychange event'i yaydı. İstekler ve konsol satırları sayfanın yaptıkları:

  0.0 s request /api/orders/1042/status
  0.2 s page: Order 1042: Paid
  5.0 s request /api/orders/1042/status
  5.2 s page: Order 1042: Paid
  7.0 s visibility: hidden
 19.0 s visibility: visible
 19.0 s request /api/orders/1042/status
 19.2 s page: Order 1042: Paid
 24.0 s request /api/orders/1042/status
 24.2 s page: Order 1042: Shipped
 29.0 s request /api/orders/1042/status
 29.2 s page: Order 1042: Shipped
 31.0 s final text: Order 1042: Shipped

[api] listening on http://localhost:5180
[api] #1 GET /api/orders/1042/status
[api] #1 200 Paid
[api] #2 GET /api/orders/1042/status
[api] #2 200 Paid
[api] #3 GET /api/orders/1042/status
[api] #3 200 Paid
[api] #4 GET /api/orders/1042/status
[api] #4 200 Shipped
[api] #5 GET /api/orders/1042/status
[api] #5 200 Shipped

31 saniyede yedi yerine beş istek, sayfa gizliyken de hiç istek yok. Burada startWith önemli: o olmasa akış ilk visibilitychange'i beklerdi ve görünür açılan bir sayfa hiç polling yapmazdı. Angular'da aynı pipe'ı bir service'e koyup sonucu template'e toSignal ile verirsin; o kısım bu serinin Angular yazısında.

Scheduler'lar: Değer Ne Zaman Teslim Edilir?

Zamanla çalışan her operator'ün gözden kaçması kolay bir son parametresi var: scheduler. Scheduler, işin ne zaman ve event loop'un hangi aşamasında çalışacağına karar veriyor. Event loop'un kendisi JavaScript İleri serisinin "JavaScript Event Loop Derinlemesine" yazısında; burada RxJS'in dört scheduler'ının oraya nasıl yerleştiğine, TestScheduler olmadan, gerçek zamanda bakıyoruz:

import { asapScheduler, asyncScheduler, observeOn, of, queueScheduler, subscribeOn } from "rxjs";

// gerçek zaman, TestScheduler yok: satırlar hangi sırayla yazılır?
console.log("1. script start");
setTimeout(() => console.log("setTimeout(0)"), 0);
Promise.resolve().then(() => console.log("Promise.then"));

of("asyncScheduler").pipe(observeOn(asyncScheduler)).subscribe((v) => console.log(`observeOn ${v}`));
of("asapScheduler").pipe(observeOn(asapScheduler)).subscribe((v) => console.log(`observeOn ${v}`));
of("queueScheduler").pipe(observeOn(queueScheduler)).subscribe((v) => console.log(`observeOn ${v}`));
of("no scheduler").subscribe((v) => console.log(v));
of("subscribeOn asyncScheduler").pipe(subscribeOn(asyncScheduler)).subscribe((v) => console.log(v));

console.log("2. script end");

1. script start
observeOn queueScheduler
no scheduler
2. script end
Promise.then
observeOn asapScheduler
setTimeout(0)
observeOn asyncScheduler
subscribeOn asyncScheduler

queueScheduler da, hiç scheduler vermemek de senkron çalıştı, "script end"ten önce. queueScheduler'ın farkı ancak bir iş kendi içinden yeni iş planladığında ortaya çıkıyor: iç içe işi recursive çalıştırmak yerine sıraya koyuyor ve call stack'i düz tutuyor. asapScheduler bir microtask olarak çalıştı: kendinden önce sıraya giren Promise.then'in hemen ardından ve bütün timer'lardan önce; RxJS 7.8.2'de Promise.resolve().then üzerine kurulu. asyncScheduler ise bir timer (RxJS 7.8.2 içeride setInterval kullanıyor), bu yüzden setTimeout(0)'dan sonra çalıştı. debounceTime, throttleTime, timer, interval ve delay'in varsayılan scheduler'ı da bu; TestScheduler'ın onu sanal zamanla değiştirebilmesi de bu sayede.

observeOn ve subscribeOn farklı şeyleri taşıyor. observeOn(scheduler) o noktadan sonraki her bildirimi (next, error, complete) scheduler üzerinden teslim ediyor. subscribeOn(scheduler) ise subscribe işleminin kendisini geciktiriyor, yani kaynak üretmeye daha geç başlıyor; son satır timer'dan sonra yazıldı, çünkü of'un tamamı asyncScheduler'da başladı. Uygulama kodunda ikisine de nadiren ihtiyaç duyarsın; asıl kullanım alanları ağır senkron işi mevcut task'tan dışarı taşımak ya da UI işini frame'lerle hizalamak. Bu da bizi dördüncü scheduler'a getiriyor.

animationFrameScheduler: Frame Başına Bir Güncelleme

animationFrameScheduler, işi tarayıcı ekrana çizmeden hemen önce requestAnimationFrame ile çalıştırıyor. Klasik kullanımı scroll handler'ı: Bookstore'un uzun katalog sayfasında bir ilerleme çubuğu ve "Back to top" butonu scroll konumunu takip ediyor. auditTime(0, animationFrameScheduler), "bir event'ten sonra bir sonraki frame'e kadar bekle, sonra son değeri al" demek:

fromEvent(window, "scroll", { passive: true })
  .pipe(
    tap(() => scrollEvents++),
    auditTime(0, animationFrameScheduler), // frame başına en fazla bir güncelleme, son konumla
    map(() => window.scrollY),
  )
  .subscribe((y) => {
    scrollUpdates++;
    progress.value = y / (document.documentElement.scrollHeight - window.innerHeight);
    toTop.hidden = y < 600;
    scrollStats.textContent = `scroll events: ${scrollEvents}, screen updates: ${scrollUpdates}`;
  });

Sayfa (web/catalog-scroll.html) hem scroll event'lerini hem de ekran güncellemelerini sayıyor. Playwright ile önce 30 adım mouse wheel, sonra tek bir task içinde 200 scrollBy çağrısı, ardından "Back to top"a tıklama yaptık:

30 mouse wheel steps     -> scroll events: 28, screen updates: 28 | Back to top visible: true
200 scrollBy in one task -> scroll events: 29, screen updates: 29
click Back to top        -> scroll events: 30, screen updates: 30 | scrollY: 0 | Back to top visible: false
200 stock messages       -> stock messages: 200, screen updates: 1 | Dune stock: 12
Chromium 158.0.8127.18

İlk üç satırda bir sürpriz var: scroll event'leri ile ekran güncellemeleri eşit, 200 scrollBy çağrısı ise tek bir scroll event'i üretti. Tarayıcı scroll'u rendering adımlarının parçası olarak zaten frame başına en fazla bir kez yayıyor; yani Chromium'da auditTime(0, animationFrameScheduler) burada scroll güncellemelerinin sayısını azaltmıyor. Yine de verdiği bir şey var: DOM'a (sayfanın tarayıcı belleğindeki element ağacı) yazma, frame callback'inde, çizimden hemen önce oluyor ve handler bir gün frame'e bağlı olmayan bir kaynakla çalışırsa pipe yine doğru kalıyor.

Son satır böyle bir kaynağı gösteriyor. WebSocket'ten gelen stok mesajları tek bir frame içinde defalarca gelebilir; sayfa bunlardan 200 tanesini tek bir task'ta taklit ediyor:

fromEvent(stockFeed, "stock")
  .pipe(
    tap(() => stockMessages++),
    map((event) => (event as CustomEvent<number>).detail),
    auditTime(0, animationFrameScheduler),
  )
  .subscribe((count) => {
    stockUpdates++;
    stock.textContent = `Dune stock: ${count}`;
    stockStats.textContent = `stock messages: ${stockMessages}, screen updates: ${stockUpdates}`;
  });

fromEvent(document.querySelector("#simulate")!, "click").subscribe(() => {
  for (let i = 1; i <= 200; i++) stockFeed.dispatchEvent(new CustomEvent("stock", { detail: 212 - i }));
});

200 mesaj, bir ekran güncellemesi ve ekranda son değer (12). DOM'u 200 kez güncellemek aynı görüntüyü üretirdi, çünkü kullanıcı yalnız çizileni görüyor. İki durumdan çıkan ders şu: bir event kaynağını yavaşlatmadan önce gerçekte ne yaptığını ölç; animationFrameScheduler'ı tek frame içinde çok değer gelebilen yerlerde kullan.

Üretimde Dikkat

Süreleri alışkanlıkla değil ölçümle seç. Arama kutusu için 300 ms'lik debounce iyi bir başlangıç; otomatik kaydetmede 800 ms ile 2 saniye arası daha yaygın, çünkü her kayıt bir yazma işlemi. Timeout için endpoint'in gerçek gecikme dağılımına bak (ör. APM (Application Performance Monitoring; uygulamanın performansını ve hatalarını izleyen araç) aracındaki 99. persentil) ve timeout'u bunun biraz üstüne koy; normal gecikmenin altındaki bir timeout, yavaş bir API'yi bozuk bir API'ye çevirir.

Polling, açık sekme sayısıyla çarpılan bir yük problemi. Sipariş sayfası açık on bin müşteri ve beş saniyelik periyot, nadiren değişen bir sayfa için saniyede iki bin istek demek. Gizliyken durmak, son durumdan sonra durmak ve eski siparişler için daha uzun bir periyot bu yükün çoğunu kesiyor. Durum gerçekten anında gelmeliyse sunucudan push daha doğru araç; seçenekleri Real-time Uygulamalar serisinin "Real-time Seçenekleri: Polling, Long Polling, Server-Sent Events, WebSocket ve SignalR — Hangisi Ne Zaman?" yazısı karşılaştırıyor.

Son olarak, tarayıcılar gizli sekmelerdeki timer'ları yavaşlatıyor. Chrome ekibinin anlattığına göre Chrome 88'den beri gizli bir sayfadaki timer'lar saniyede bir kontrol ediliyor, beş dakikadan uzun süredir gizli olan bir sayfada art arda çalışan timer'lar ise dakikada bir. Gizli bir sekmede interval'ının zamanında çalışacağının garantisi yok; polling'i tarayıcıya bırakmak yerine açıkça durdurmanın bir sebebi daha.

Sık Düşülen Tuzaklar

• mergeMap ile polling yapmak. Yavaş bir istek bir sonrakiyle üst üste biner, cevaplar sırasız gelir ve ekrandaki durum geriye gider. exhaustMap ya da repeat({ delay }) kullan.
• Arama kutusunun dış akışına timeout koymak. İsteği değil kullanıcının duraklamasını ölçer ve kutuyu öldürür; switchMap'in içine koy.
• Otomatik kaydetmeyi debounceTime'dan sonra gelen bir takeUntil ile bitirmek. Bekleyen düzenleme atılır; kaynağı tamamla ya da component'ten uzun yaşayan bir service'ten kaydet.
• Son değer gerektiğinde throttleTime kullanmak. Varsayılanı { leading: true, trailing: false }, yani bir serinin son değeri kaybolur; auditTime ya da debounceTime kullan.
• Her bufferTime sonucunu sunucuya göndermek. Boş periyotlar boş array üretir; onları süz.
• auditTime(0, animationFrameScheduler)'ın scroll event'lerini azaltmasını beklemek. Tarayıcı onları zaten frame başına bir kez veriyor; kazanç, frame'e bağlı olmayan kaynaklarda.

Küçük Bir Deneme

★☆☆ 01-predict.ts'te debounceTime(500) yerine önce auditTime(500), sonra throttleTime(500) koy. Çalıştırmadan önce kaydedilecek metinleri tahmin et. Hangisini otomatik kaydetmede asla kullanmazdın ve neden?

★★☆ 03-autosave.ts'e bir "Save now" butonu ekle: tıklama o anki metni hemen kaydetsin ve aynı metin için bekleyen debounce'lu kayıt ardından bir daha çalışmasın. İpucu: tıklama akışını distinctUntilChanged'den önce debounce'lu akışla merge et.

★★★ web/order-status.ts'i, durum değişmedikçe polling periyodu büyüyecek (5, 10, 20 saniye, en fazla 60) ve bir değişiklikten sonra 5 saniyeye dönecek şekilde değiştir. repeat({ delay: (count) => timer(…) }) ya da expand kullan ve aralıkların gerçekten büyüdüğünü sunucu logundan kontrol et.

Bu yazıda bir akışı bekletmeyi, atlatmayı ve tekrarlatmayı gördük. Bir sonraki yazıda değerleri kendimiz üreteceğiz: Subject, BehaviorSubject, ReplaySubject ve AsyncSubject'in geç gelen subscriber'a ne verdiğini, public bırakılan bir Subject'in neden sorun olduğunu ve Angular 22'de state'in neden signal'da, event'lerin ise neden Subject'te kalabileceğini göreceğiz.