RxJS'te Stream'leri Birleştirmek: combineLatest, withLatestFrom, forkJoin, zip, merge ve concat

Sürüm notu: Bu yazıdaki örnekler RxJS 7.8.2, TypeScript 6.0 ve Node.js 24 ile denenmiştir; zamanlı örnekler RxJS'in TestScheduler'ı ile sanal zamanda çalıştırılmıştır.

Önce Tahmin Et

Sitedeki serilerin ortak örnek uygulaması Bookstore'un (bir online kitapçı) katalog sayfasında üç filtre var: kategori, sıralama ve fiyat aralığı. Üçünü combineLatest ile tek bir sorguya çeviriyoruz. Kullanıcı 100. ms'de kategori olarak "classics", 300. ms'de sıralama olarak "price-asc" seçiyor; fiyat aralığı kaydırıcısına hiç dokunmuyor:

import { combineLatest, NEVER, timer, type Observable } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import type { PriceRange } from "./catalog-api.ts";

runInVirtualTime("catalog filters", () => {
  const category$ = emitAt([100, "classics"]);
  const sort$ = emitAt([300, "price-asc"]);
  const priceRange$: Observable<PriceRange> = NEVER; // kullanıcı fiyat kaydırıcısına hiç dokunmuyor

  combineLatest({ category: category$, sort: sort$, priceRange: priceRange$ }).subscribe((filters) =>
    log(`query: ${JSON.stringify(filters)}`),
  );
  timer(1000).subscribe(() => log("one second later"));
});

Bir saniye içinde kaç "query" satırı görürsün? Bir, iki, yoksa hiç?

Önceki yazıda tek bir akışın her değerinden yeni bir akış ürettik. Bu yazıda birbirinden bağımsız akışları birleştiriyoruz: filtreler, butonlar, birden çok API çağrısı. RxJS'te bunun için yedi temel araç var: combineLatest, withLatestFrom, forkJoin, zip, merge, concat ve race. Hepsi "iki akışı birleştir" gibi görünüyor, ama her biri farklı bir soruya cevap veriyor. Örnekler önceki yazıdaki sanal zaman yardımcılarını (runInVirtualTime, emitAt, log) ve sahte HTTP çağrısını (fakeRequest) kullanıyor.

Her Birleştirici Bir Soruya Cevap Verir

Aynı iki kaynakla marble'ları yan yana görmek farkı en hızlı anlatan yol. İlk kaynak harf (ör. kategori değişiklikleri), ikincisi rakam (ör. sayfa değişiklikleri) yayıyor; çıktıdaki a1 gibi değerler iki kaynağın o anda eşleşen değerleri:

import { combineLatest, concat, map, merge, race, withLatestFrom, zip } from "rxjs";
import { showMarbles } from "../../tools/marble.ts";

// iki kaynak: harfler (ör. kategori değişiklikleri) ve rakamlar (ör. sayfa değişiklikleri)
const letters = "-a----b---c--|";
const digits = "---1-----2----|";
const pair = ([x, y]: [string, string]) => x + y;

showMarbles("combineLatest", (cold) => combineLatest([cold<string>(letters), cold<string>(digits)]).pipe(map(pair)));
showMarbles("withLatestFrom", (cold) => cold<string>(letters).pipe(withLatestFrom(cold<string>(digits)), map(pair)));
showMarbles("zip", (cold) => zip([cold<string>(letters), cold<string>(digits)]).pipe(map(pair)));
showMarbles("merge", (cold) => merge(cold<string>(letters), cold<string>(digits)));
showMarbles("concat", (cold) => concat(cold<string>("-a--b|"), cold<string>("-1--2|")));
showMarbles("race", (cold) => race(cold<string>("---a--b|"), cold<string>("-1--2---|")));

combineLatest
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: ---a--b--cd---|  a=a1 b=b1 c=b2 d=c2
withLatestFrom
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: ------a---b--|   a=b1 b=c2
zip
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: ---a-----b----|  a=a1 b=b2
merge
  in : -a----b---c--|   
  in : ---1-----2----|  
  out: -a-1--b--2c---|  
concat
  in : -a--b|           
  in : -1--2|           
  out: -a--b-1--2|      
race
  in : ---a--b|         
  in : -1--2---|        
  out: -1--2---|        

İlk üç satır en sık karıştırılanlar. combineLatest ilk değerini ancak 3. frame'de, rakam kaynağı ilk kez yayınca verdi; ondan önce gelen a tek başına hiçbir şey üretmedi. Sonra her yeni harf ya da rakamda o anki ikiliyi yaydı (a1, b1, b2, c2). withLatestFrom yalnız harf geldiğinde yaydı ve rakam kaynağı henüz yaymamışken gelen a'yı sessizce yuttu; rakamların kendisi hiçbir şeyi tetiklemiyor. zip ise değerleri sıra numarasına göre eşleştirdi: birinci harf birinci rakamla (a1), ikinci harf ikinci rakamla (b2); c'nin eşi hiç gelmedi.

combineLatest ve startWith: Katalog Filtreleri

Tahminin cevabı:

--- catalog filters
 1000 ms  one second later

Hiç sorgu çıkmadı. combineLatest, her kaynaktan en az bir değer görmeden hiçbir şey yaymıyor; fiyat aralığı hiç yaymadığı için kullanıcı ne seçerse seçsin liste hiç yüklenmez. Hata da yok, uyarı da: sayfa sadece boş kalıyor. Gerçek uygulamada form kontrolleri ve route parametreleri çoğu zaman tam bu şekilde, kullanıcı dokunana kadar değer yaymayan kaynaklar.

Çözüm her filtreye bir başlangıç değeri vermek. startWith operator'ü, kaynağın önüne subscribe anında senkron yayılan bir değer ekliyor. Snapshot'taki src/05/catalog-api.ts, Bookstore'un katalog endpoint'ini taklit ediyor: ortak örnek verideki altı kitabı kategori, fiyat ve sıralamaya göre süzüp 200 ms sonra dönüyor. Filtreleri dictionary imzasıyla birleştiriyoruz; sonuç, alan adlarıyla gelen tek bir object:

import { combineLatest, distinctUntilChanged, map, startWith, switchMap } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { getBooks, toQueryString, type CatalogQuery, type PriceRange, type SortOrder } from "./catalog-api.ts";

runInVirtualTime("combineLatest + startWith", () => {
  // kullanıcının katalog sayfasında ne yaptığı ve ne zaman
  const category$ = emitAt<CatalogQuery["category"]>([100, "classics"]);
  const priceRange$ = emitAt<PriceRange>([500, { min: 0, max: 350 }]);
  const sort$ = emitAt<SortOrder>([900, "price-asc"]);

  const query$ = combineLatest({
    category: category$.pipe(startWith<CatalogQuery["category"]>("all")),
    priceRange: priceRange$.pipe(startWith({ min: 0, max: 1000 })),
    sort: sort$.pipe(startWith<SortOrder>("title")),
  });

  query$
    .pipe(
      distinctUntilChanged((a, b) => toQueryString(a) === toQueryString(b)),
      switchMap((query) => getBooks(query)),
      map((titles) => titles.join(", ")),
    )
    .subscribe((list) => log(`screen: ${list}`));
});

--- combineLatest + startWith
    0 ms  -> GET /books?category=all&min=0&max=1000&sort=title
  100 ms  x  GET /books?category=all&min=0&max=1000&sort=title cancelled
  100 ms  -> GET /books?category=classics&min=0&max=1000&sort=title
  300 ms  <- GET /books?category=classics&min=0&max=1000&sort=title
  300 ms  screen: Emma 320, Frankenstein 280, Middlemarch 380, Pride and Prejudice 300
  500 ms  -> GET /books?category=classics&min=0&max=350&sort=title
  700 ms  <- GET /books?category=classics&min=0&max=350&sort=title
  700 ms  screen: Emma 320, Frankenstein 280, Pride and Prejudice 300
  900 ms  -> GET /books?category=classics&min=0&max=350&sort=price-asc
 1100 ms  <- GET /books?category=classics&min=0&max=350&sort=price-asc
 1100 ms  screen: Frankenstein 280, Pride and Prejudice 300, Emma 320

Sayfa açılır açılmaz (0. ms) varsayılan filtrelerle ilk sorgu gitti; kullanıcı 100. ms'de kategoriyi değiştirince switchMap o isteği iptal edip yenisini attı. Sonraki iki değişiklik de birer sorguya dönüştü ve son ekranda klasikler 350 TL altında, fiyata göre artan sırada: Frankenstein 280, Pride and Prejudice 300, Emma 320. Zincir, önceki iki yazının parçalarından oluşuyor: distinctUntilChanged query string üzerinden karşılaştırıyor (üçüncü yazıdaki object tuzağı), switchMap eski sorguyu iptal ediyor (dördüncü yazı).

Dictionary imzası (combineLatest({ category, priceRange, sort })) array imzasına göre daha okunur: değerlere filters[0] yerine filters.category diye erişiyorsun ve yeni bir filtre eklemek sıralamayı bozmuyor. RxJS 7'de combineLatest(a$, b$) gibi rest parametreli imza ise deprecated; type tanımında bir sonraki major sürümde kaldırılacağı yazıyor. Array ya da object ver.

withLatestFrom: Tetikleyici ve Bağlam

Bookstore'un kurumsal sipariş ekranında okul, kütüphane ya da şirket adına alım yapan kullanıcı adet ve kurum bilgilerini doldurup "Request Quote" butonuna basıyor. Burada sorgunun her alan değişikliğinde değil, yalnız butona basınca gitmesi gerekiyor. Aynı akışları withLatestFrom ve combineLatest ile karşılaştıralım:

import { combineLatest, map, startWith, withLatestFrom } from "rxjs";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";

runInVirtualTime("withLatestFrom: the click decides, the form only supplies context", () => {
  // kurumsal sipariş formu: alıcı alanları düzenliyor, sonra "Request Quote"a basıyor
  const quantity$ = emitAt([100, 20], [200, 25], [600, 30]).pipe(startWith(10));
  const institution$ = emitAt([150, "Hillside School"]);
  const quoteClicks$ = emitAt([50, "click"], [400, "click"], [800, "click"]);

  quoteClicks$
    .pipe(
      withLatestFrom(quantity$, institution$),
      map(([, quantity, institution]) => `quote: ${quantity} x Dune for ${institution}`),
    )
    .subscribe((line) => log(line));
});

runInVirtualTime("combineLatest: every field change fires", () => {
  const quantity$ = emitAt([100, 20], [200, 25], [600, 30]).pipe(startWith(10));
  const institution$ = emitAt([150, "Hillside School"]);

  combineLatest([quantity$, institution$])
    .pipe(map(([quantity, institution]) => `quote: ${quantity} x Dune for ${institution}`))
    .subscribe((line) => log(line));
});

--- withLatestFrom: the click decides, the form only supplies context
  400 ms  quote: 25 x Dune for Hillside School
  800 ms  quote: 30 x Dune for Hillside School
--- combineLatest: every field change fires
  150 ms  quote: 20 x Dune for Hillside School
  200 ms  quote: 25 x Dune for Hillside School
  600 ms  quote: 30 x Dune for Hillside School

withLatestFrom ile teklif yalnız tıklamalarda istendi ve her seferinde o anki adedi kullandı: 400. ms'de 25, 800. ms'de 30. combineLatest ise her alan değişikliğinde bir teklif üretti; bu bir arama listesi için doğru davranış, ama bir teklif isteği için gereksiz trafik ve belki de kurum adına yanlışlıkla gönderilmiş bir talep.

Çıktıda bir şey eksik: 50. ms'deki tıklama hiçbir şey üretmedi. O anda institution$ henüz bir değer yaymamıştı ve withLatestFrom, bağlam kaynaklarından biri yaymadıysa tetikleyiciyi sessizce atıyor. Kullanıcı açısından bu "butona bastım, hiçbir şey olmadı" demek. Bağlam kaynaklarına startWith ile anlamlı bir başlangıç değeri ver ya da butonu, gerekli alanlar dolana kadar devre dışı bırak. İki operator'ün iş bölümü kısaca şu: combineLatest'te bütün kaynaklar eşit, herhangi biri tetikler; withLatestFrom'da bir kaynak tetikler, diğerleri yalnız bağlam verir.

forkJoin: Hepsi Bitince Bir Kez

Sipariş detay sayfası üç bilgiye ihtiyaç duyuyor: sipariş, müşteri ve kargo durumu. Üç istek birbirinden bağımsız, dolayısıyla sırayla değil paralel gitmeli. forkJoin hepsine aynı anda subscribe oluyor, hepsi tamamlanınca her birinin son değerini tek seferde veriyor. Ortak örnek verideki 1042 numaralı sipariş (Ada, Dune ve Emma) için:

import { EMPTY, forkJoin, interval, map, timer } from "rxjs";
import { customers, orders } from "../../data/books.ts";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

const order = orders[0]; // 1042 numaralı sipariş: Ada, Dune + Emma

runInVirtualTime("forkJoin: three calls in parallel, one result", () => {
  forkJoin({
    order: fakeRequest(`GET /orders/${order.id}`, 300, () => order),
    customer: fakeRequest(`GET /customers/${order.customerId}`, 100, () => customers[0]),
    shipment: fakeRequest(`GET /orders/${order.id}/shipment`, 200, () => ({ status: "Shipped", trackingNo: "TRK-1042" })),
  }).subscribe(({ order, customer, shipment }) =>
    log(`order ${order.id} for ${customer.name}: ${order.items.length} items, ${order.status}, shipment ${shipment.status}`),
  );
});

runInVirtualTime("forkJoin with a source that never completes", () => {
  const liveStock$ = interval(1000).pipe(map((i) => 12 - i)); // stok güncellemeleri, hiç tamamlanmaz
  forkJoin({
    order: fakeRequest(`GET /orders/${order.id}`, 300, () => order),
    stock: liveStock$,
  }).subscribe(() => log("never printed"));
  timer(5000).subscribe(() => log("still waiting after 5 seconds"));
});

runInVirtualTime("forkJoin edge cases", () => {
  forkJoin([]).subscribe({
    next: (v) => log(`[] next ${JSON.stringify(v)}`),
    complete: () => log("[] complete"),
  });
  forkJoin({ order: fakeRequest(`GET /orders/${order.id}`, 300, () => order), coupons: EMPTY }).subscribe({
    next: () => log("never printed"),
    complete: () => log("{ order, coupons: EMPTY } complete without a value"),
  });
});

--- forkJoin: three calls in parallel, one result
    0 ms  -> GET /orders/1042
    0 ms  -> GET /customers/1
    0 ms  -> GET /orders/1042/shipment
  100 ms  <- GET /customers/1
  200 ms  <- GET /orders/1042/shipment
  300 ms  <- GET /orders/1042
  300 ms  order 1042 for Ada: 2 items, Paid, shipment Shipped
--- forkJoin with a source that never completes
    0 ms  -> GET /orders/1042
  300 ms  <- GET /orders/1042
 5000 ms  still waiting after 5 seconds
--- forkJoin edge cases
    0 ms  [] complete
    0 ms  -> GET /orders/1042
    0 ms  { order, coupons: EMPTY } complete without a value
    0 ms  x  GET /orders/1042 cancelled

İlk senaryoda üç istek 0. ms'de birlikte gitti ve sonuç en yavaş isteğin süresinde (300 ms) geldi; sırayla gitselerdi 600 ms sürerdi. forkJoin, Promise dünyasındaki Promise.all'ın karşılığı ve HTTP gibi tek değer yayıp tamamlanan kaynaklar için tasarlanmış.

Diğer iki senaryo bu tasarımın sınırları. Tamamlanmayan bir kaynak (canlı stok akışı, bir form'un valueChanges'i, bir Subject) forkJoin'i sonsuza kadar bekletiyor: sipariş 300. ms'de geldi, ama 5 saniye sonra hâlâ sonuç yok. Sanal zamanlı örnek de bu yüzden runInVirtualTime'ın sanal saati 60 saniyede durdurmasına güveniyor; durdurmasa interval sonsuza kadar sürerdi. Boş array ile forkJoin([]) hiç değer yaymadan hemen tamamlanıyor; "seçili ürünlerin stoklarını çek" gibi bir listede liste boşsa ekran bir sonuç bekleyip hiç güncellenmeyebilir. Bir kaynak değer yaymadan tamamlanırsa (EMPTY ya da hiçbir değeri filter'dan geçemeyen bir akış) forkJoin da değer yaymadan tamamlanıyor ve diğer istekleri iptal ediyor: son satırdaki x, sipariş isteğinin boşa gittiğini gösteriyor. Bir kaynağın hata vermesi de aynı şekilde bütün forkJoin'i hataya düşürüp diğerlerini iptal eder; kısmi sonuçla devam etmek istiyorsan her kaynağa kendi hata yönetimini eklemen gerekir (bu serinin error handling yazısında).

zip, merge, concat ve race

Marble'lardaki son dört satıra dönelim. merge, birden çok kaynağın değerlerini geldikleri sırayla tek akışa koyuyor; hiçbir eşleştirme yapmıyor. Bookstore'da "sepete eklendi" event'i hem katalog listesinden hem de kitap detay sayfasından geliyorsa, ikisini merge ile tek bir akışa almak sepet sayacını güncellemek için yeterli. merge bütün kaynaklar tamamlanınca tamamlanır; içlerinden biri hata verirse hepsi biter.

concat kaynakları sırayla çalıştırıyor: ikinci kaynağa ancak ilki tamamlanınca subscribe oluyor (marble'da 1 ve 2, b'den sonra ve kendi zaman çizelgesinin başından başladı). Tipik kullanımı, önce cache'teki değeri sonra taze değeri göstermek: concat(cachedBooks$, fetchBooks$). Dikkat edilecek nokta, ilk kaynak hiç tamamlanmazsa ikincisinin hiç başlamaması.

race, ilk değeri yayan kaynağı seçiyor ve diğerlerine unsubscribe oluyor: marble'da rakam kaynağı 1. frame'de yaydığı için harfler hiç görünmedi. Aynı veriyi iki kaynaktan (ör. iki depo servisinden) isteyip hızlı olanı almak ya da bir isteği bir zaman aşımıyla yarıştırmak için kullanılıyor; zaman aşımı için ise RxJS'in kendi timeout operator'ü daha açık (bu serinin zaman yazısında).

zip en az ihtiyaç duyulanı. Değerleri sıra numarasıyla eşleştiriyor, yani kaynaklardan biri hızlıysa onun değerleri eşleri gelene kadar bellekte bekliyor. Gerçekten "birinci ile birinci, ikinci ile ikinci" demek istediğin durumlar az: aynı sırayla gönderilen istek ve cevap çiftleri ya da animasyon adımları. Kullanıcı event'lerini zip'lemek neredeyse her zaman hata; iki akışın hızı farklıysa buffer sınırsız büyür.

Üretimde Dikkat

Birleştiricilerle ilgili production sorunları genelde "hiçbir şey olmuyor" şeklinde görünür, hata olarak değil: combineLatest bir kaynağı beklerken, forkJoin tamamlanmayan bir kaynağı beklerken ya da withLatestFrom tetikleyiciyi yutarken konsolda hiçbir iz kalmaz. Bu yüzden her kaynağı ayrı ayrı izlemek işe yarıyor: geliştirme sırasında kaynakların sonuna geçici bir tap({ next, complete }) ekleyip hangisinin hiç yaymadığını ya da tamamlanmadığını görmek, "liste neden yüklenmiyor" sorusunu dakikalar içinde cevaplar. Bu tür debug yardımcılarını bir custom operator'e çevirmek bu serinin custom operator yazısının konusu.

Ağ tarafında forkJoin ile combineLatest'in farkı ölçülebilir. Sipariş detay sayfasında üç isteğin Network sekmesinde aynı anda başladığını (waterfall'da üst üste) görmelisin; merdiven gibi art arda başlıyorlarsa bir yerde iç içe subscribe ya da concatMap zinciri var demektir. Bookstore'un ASP.NET Core tarafında üç ayrı endpoint yerine sipariş detayını tek bir endpoint'te toplamak da bir seçenek; istemcide forkJoin sayfa başına istek sayısını azaltmaz, yalnız bekleme süresini en yavaş isteğe indirir.

Son olarak combineLatest'te aynı kaynaktan türeyen iki akışı birleştirirsen, kaynak her değiştiğinde önce biri sonra diğeri güncellendiği için arada bir anlığına tutarsız bir ikili yayılır. Bu ara değer bir HTTP isteğine dönüşüyorsa gereksiz bir istek daha demek; filtreleri tek bir state object'inde tutmak ya da distinctUntilChanged ile query string'i karşılaştırmak bu tekrarları azaltır.

Sık Düşülen Tuzaklar

• combineLatest'e henüz değer yaymamış bir kaynak vermek. Bütün kaynaklar en az bir kez yaymadan hiçbir şey çıkmaz; form ve filtre kaynaklarına startWith ile başlangıç değeri ver.
• forkJoin'e tamamlanmayan bir kaynak vermek (canlı akış, valueChanges, Subject). Sonuç hiç gelmez; tek değer yeterliyse kaynağa take(1) ekle.
• Boş bir array'i forkJoin'e verip sonuç beklemek. forkJoin([]) değer yaymadan tamamlanır; boş listeyi önceden kontrol et ya da defaultIfEmpty([]) ekle.
• Değer yaymadan tamamlanan bir kaynağın (EMPTY, hiçbir değeri geçmeyen bir filter) bütün forkJoin'i değer üretmeden bitireceğini unutmak.
• withLatestFrom'un bağlam kaynakları henüz yaymamışken gelen tetikleyiciyi sessizce attığını bilmemek.
• Kullanıcı event'lerini zip ile eşleştirmek. Hızlı taraf bellekte birikir ve değerler yanlış eşleşir.
• Rest parametreli combineLatest(a$, b$) ve forkJoin(a$, b$) imzalarını kullanmak. RxJS 7'de deprecated; array ya da object ver.

Küçük Bir Deneme

★☆☆ 02-catalog-filters.ts'te priceRange$'tan startWith'i kaldır. İlk sorgu hangi milisaniyede gider ve kullanıcı o ana kadar ekranda ne görür? Sonra startWith'i geri koyup kategori değişikliğini 100 yerine 50. ms'ye çek; iptal edilen istek sayısı değişiyor mu?

★★☆ 03-with-latest-from.ts'te 50. ms'deki tıklamanın kaybolmasını engelle: önce institution$'a bir başlangıç değeri vererek, sonra başlangıç değeri olmadan, tıklamaları kurum seçilene kadar bekleten bir çözümle. İkinci çözümde kullanıcı aynı anda iki kez tıklarsa ne olur?

★★★ 04-fork-join.ts'teki sipariş detayını, kargo servisi hata verdiğinde sayfanın yine sipariş ve müşteri bilgisini gösterecek şekilde değiştir; kargo için "unknown" yazsın. Bunu yaparken forkJoin'in hangi davranışını aştığını ve hatayı hangi seviyede yakaladığını açıkla. İpucu: hata yakalama operator'lerini henüz görmedik, ama catchError'ın her kaynağın kendi pipe'ında durması gerektiğini tahmin edebilirsin.

Bu yazıdaki zamanlamaları hep kullanıcının ne zaman tıkladığı ya da API'nin ne kadar sürede cevap verdiği belirledi. Bir sonraki yazıda zamanı kendimiz yöneteceğiz: debounceTime, throttleTime, auditTime ve timeout arasındaki farkı, RxJS'in scheduler'larını ve sipariş durumunu sayfa gizliyken durduran bir polling'i göreceğiz.