RxJS'te Multicasting: share, shareReplay, connectable ve HTTP İsteğini Paylaşmak

Sürüm notu: Bu yazıdaki örnekler RxJS 7.8.2, Angular 22.2, TypeScript 6.0 ve Node.js 24 ile denenmiştir (Angular projesi, Angular 22.2'nin desteklediği TypeScript 6.0'ı kullanıyor); zamanlı örnekler RxJS'in TestScheduler'ı ile sanal zamanda, Angular örneği Playwright 1.65 ile headless Chromium'da çalıştırılmıştır.

Önce Tahmin Et

Sitedeki serilerin ortak örnek uygulaması Bookstore'un (bir online kitapçı) katalog sayfası kategori listesine üç yerde ihtiyaç duyuyor: menü, filtre ve footer. Üç istek atmamak için isteğe share() ekliyoruz. Menü ve filtre hemen, footer bir saniye sonra subscribe oluyor:

import { share, timer } from "rxjs";
import { fromFetch } from "rxjs/fetch";

// GET /api/categories bir kez, bütün subscriber'lar paylaşıyor... öyle mi?
const categories$ = fromFetch("http://localhost:5180/api/categories", {
  selector: (response) => response.json() as Promise<string[]>,
}).pipe(share());

categories$.subscribe((c) => console.log("menu:", c.join(", ")));
categories$.subscribe((c) => console.log("filter:", c.join(", ")));
// footer bir saniye sonra render ediliyor
timer(1000).subscribe(() => categories$.subscribe((c) => console.log("footer:", c.join(", "))));

Sahte API çalışırken (node api/server.ts) sunucu logunda kaç GET /api/categories satırı görürsün: bir, iki, yoksa üç?

Bu serinin ilk yazısında cold bir Observable'a yapılan her subscribe'ın producer'ı yeniden çalıştırdığını ve aynı HTTP Observable'ına iki subscribe'ın iki istek attığını görmüştük. Önceki yazıda ise bir Subject, kendi kodumuzun ürettiği değerleri paylaştı. Bu yazı ikisini birleştiriyor: var olan cold bir Observable'ı share, shareReplay, connectable ve connect ile paylaşacak, gerçek bir Angular 22 projesinde ölçecek ve iki klasik multicasting hatasına bakacağız: hiç durmayan bir kaynak ve sonsuza kadar cache'lenen başarısız bir yanıt.

Unicast ve Multicast: İki async Pipe, İki İstek

Önce cevap:

menu: science-fiction, classics, fantasy
filter: science-fiction, classics, fantasy
footer: science-fiction, classics, fantasy

[api] listening on http://localhost:5180
[api] #1 GET /api/categories
[api] #1 200
[api] #2 GET /api/categories
[api] #2 200

İki istek. Menü ve filtre tek bir isteği paylaştı, ama footer ikinci bir istek başlattı. share(), kaynak tamamlanınca her şeyi unutuyor: footer subscribe olduğunda ilk istek bitmişti ve tamamlanmadan sonra gelen yeni bir subscriber kaynağı yeniden başlatıyor. share bir sonucu değil, süren bir çalışmayı paylaşıyor. Bunu birazdan düzelteceğiz; önce aynı sorunu, çoğu zaman kimse subscribe yazmadan ortaya çıktığı yerde, Angular'da görelim.

Bir Observable varsayılan olarak unicast: her subscriber producer'ın kendine ait bir çalışmasını alıyor. Subject ise multicast: tek çalışma, çok subscriber. Multicasting operator'leri kaynak ile subscriber'lar arasına bir Subject koyuyor. Angular'ın HttpClient'ı cold, unicast bir Observable döndürüyor; bunu dokümana bakıp söylemek kolay, biz ölçtük. bookstore-web'in katalog sayfası (snapshot'ın bookstore-web/ klasöründe) kategorileri iki kez gösteriyor, link olarak ve bir select kutusunda; ilk sürümü onları doğrudan HttpClient'tan okuyordu:

    <nav class="categories">
      @for (category of categories$ | async; track category) {
        <a href="#">{{ category }}</a>
      }
    </nav>
    <label>
      Category
      <select>
        @for (category of categories$ | async; track category) {
          <option>{{ category }}</option>
        }
      </select>
    </label>

protected readonly categories$ = this.#http.get<string[]>(`${apiUrl}/categories`);

Kataloğu headless Chromium'da açtık, sepet sayfasına gidip geri döndük; sayfanın gönderdiği istekler:

open catalog
  request /api/books
  request /api/categories
  request /api/categories
  nav: science-fiction, classics, fantasy | select: 3 options
go to cart
back to catalog
  request /api/books
  request /api/categories
  request /api/categories

İki async pipe, iki subscription, aynı liste için iki istek ve kullanıcı sayfaya döndüğünde iki istek daha. Template'te subscription'a benzeyen hiçbir şey yok; bu hatanın bu kadar yaygın olmasının sebebi de bu. Bir component aynı HttpClient Observable'ını template'te async ile, kodda subscribe ile okuduğunda ya da iki ayrı toSignal çağrısıyla kullandığında da aynısı oluyor.

share ve shareReplay Aynı Şey Değil

İki operator de aynı mekanizmanın farklı ayarları. share dört seçenek alıyor: connector (araya hangi Subject'in konacağı, varsayılanı düz bir Subject) ve paylaşılan çalışmanın ne zaman unutulacağını söyleyen üç reset kuralı: resetOnError, resetOnComplete ve resetOnRefCountZero (son subscriber ayrıldığında). RxJS 7.8.2'de shareReplay(config) aslında bir share çağrısı: connector'ı ReplaySubject, resetOnError: true, resetOnComplete: false ve resetOnRefCountZero senin verdiğin refCount değeri:

Aynı istek ve aynı üç subscriber (0, 0 ve 1000 ms) bunun ne anlama geldiğini gösteriyor; son senaryo, share'in seçenekleriyle 60 saniye sonra süresi dolan bir cache kuruyor:

import { ReplaySubject, share, shareReplay, timer, type Observable } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

const getCategories = () => fakeRequest("GET /categories", 200, () => ["science-fiction", "classics", "fantasy"]);

// sayfanın üç parçası 0 ms, 0 ms ve 1000 ms'de subscribe oluyor (dördüncüsü 90. sn'de)
function subscribeLikeAPage(categories$: Observable<string[]>, times = [0, 0, 1000]) {
  times.forEach((ms, i) => timer(ms).subscribe(() => categories$.subscribe((c) => log(`subscriber ${i + 1}: ${c.length} categories`))));
}

runInVirtualTime("no sharing", () => subscribeLikeAPage(getCategories()));
runInVirtualTime("share()", () => subscribeLikeAPage(getCategories().pipe(share())));
runInVirtualTime("shareReplay(1)", () => subscribeLikeAPage(getCategories().pipe(shareReplay(1))));
runInVirtualTime("share as a 60-second cache", () =>
  subscribeLikeAPage(
    getCategories().pipe(
      share({
        connector: () => new ReplaySubject(1),
        resetOnComplete: () => timer(60_000), // cevabı geldikten 60 sn sonra unut
        resetOnError: true,
        resetOnRefCountZero: false,
      }),
    ),
    [0, 0, 1000, 90_000],
  ),
  120_000, // sanal saat 90 sn'yi geçsin
);

--- no sharing
    0 ms  -> GET /categories
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  <- GET /categories
  200 ms  subscriber 2: 3 categories
 1000 ms  -> GET /categories
 1200 ms  <- GET /categories
 1200 ms  subscriber 3: 3 categories
--- share()
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  subscriber 2: 3 categories
 1000 ms  -> GET /categories
 1200 ms  <- GET /categories
 1200 ms  subscriber 3: 3 categories
--- shareReplay(1)
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  subscriber 2: 3 categories
 1000 ms  subscriber 3: 3 categories
--- share as a 60-second cache
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  subscriber 2: 3 categories
 1000 ms  subscriber 3: 3 categories
90000 ms  -> GET /categories
90200 ms  <- GET /categories
90200 ms  subscriber 4: 3 categories

Paylaşım olmadan üç subscriber üç istek attı. share() aynı anda gelen iki subscriber'ı tek isteğe topladı, ama geç gelen yeni bir istek başlattı, tıpkı tahmin örneğindeki gibi. shareReplay(1) tamamlanınca reset olmadı; 1000. ms'deki subscriber saklanan cevabı istek atmadan ReplaySubject'ten aldı. Son senaryo ömrü olan bir cache: resetOnComplete: () => timer(60_000) cevaptan sonra 60 saniye bekleyip reset ediyor; 1000. ms'deki subscriber bellekten, 90. saniyedeki ise yeni bir istekle cevap aldı. resetOnError: true başarısız bir isteğin saklanmamasını, resetOnRefCountZero: false ise bütün subscriber'lar ayrılsa bile süren isteğin bitmesini sağlıyor. Reset seçenekleri bir boolean ya da Observable döndüren bir fonksiyon alıyor; ikincisinde reset, o Observable yaydığında oluyor.

Kategori Listesini Uygulama Boyunca Paylaşmak

Kategoriler nadiren değişiyor, uygulama oturumu başına tek istek yeterli. Liste bir service'e taşınıyor ve paylaşılan Observable bir kez, service'in field'ında oluşturuluyor:

import { HttpClient } from '@angular/common/http';
import { inject, Service } from '@angular/core';
import { shareReplay } from 'rxjs';
import { apiUrl } from '../api';

@Service()
export class CategoryService {
  // bütün uygulama için tek istek; sonraki her subscriber cache'teki listeyi alır
  readonly categories$ = inject(HttpClient)
    .get<string[]>(`${apiUrl}/categories`)
    .pipe(shareReplay({ bufferSize: 1, refCount: false }));
}

Katalog sayfası artık categories$ = inject(CategoryService).categories$ okuyor; iki async pipe'lı template değişmedi. Öncekiyle aynı gezinti:

open catalog
  request /api/books
  request /api/categories
  nav: science-fiction, classics, fantasy | select: 3 options
go to cart
back to catalog
  request /api/books

Bütün ziyaret boyunca tek bir kategori isteği, sepetten döndükten sonra da. Kitap listesi ise yeniden isteniyor, çünkü sayfa onu kendi field'ında toSignal ile oluşturuyor; bir listenin cache'lenip cache'lenmeyeceği varsayılan değil, endpoint başına verilen bir karar. shareReplay'in nereye konduğu da operator kadar önemli: bir method içindeki shareReplay (getCategories() { return http.get(…).pipe(shareReplay(1)); }) her çağrıda yeni bir paylaşılan Observable oluşturuyor ve hiçbir şey paylaşmıyor. Paylaşılan instance, subscriber'larından uzun yaşayan bir field'da ya da değişkende saklanmalı.

refCount: Hiç Durmayan Kaynak

HTTP isteğinde refCount'un pek önemi yok, çünkü kaynak kendiliğinden tamamlanıyor ve shareReplay tamamlanınca hiç reset olmuyor. Önemli olduğu yer tamamlanmayan kaynaklar. Kitap detay sayfası Dune'un canlı stokunu shareReplay üzerinden gösteriyor; kullanıcı 2,5. saniyede sayfadan çıkıyor:

import { interval, map, shareReplay, tap, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// Dune için hiç tamamlanmayan canlı bir stok akışı
const liveStock = () =>
  interval(1000).pipe(
    map((i) => 12 - i),
    tap((stock) => log(`  feed produced ${stock}`)),
  );

for (const refCount of [false, true]) {
  runInVirtualTime(`shareReplay({ bufferSize: 1, refCount: ${refCount} })`, () => {
    const stock$ = liveStock().pipe(shareReplay({ bufferSize: 1, refCount }));
    const detailPage = stock$.subscribe((s) => log(`detail page: ${s}`));
    timer(2500).subscribe(() => {
      detailPage.unsubscribe();
      log("user leaves the detail page");
    });
  }, 5000);
}

--- shareReplay({ bufferSize: 1, refCount: false })
 1000 ms    feed produced 12
 1000 ms  detail page: 12
 2000 ms    feed produced 11
 2000 ms  detail page: 11
 2500 ms  user leaves the detail page
 3000 ms    feed produced 10
 4000 ms    feed produced 9
 5000 ms    feed produced 8
--- shareReplay({ bufferSize: 1, refCount: true })
 1000 ms    feed produced 12
 1000 ms  detail page: 12
 2000 ms    feed produced 11
 2000 ms  detail page: 11
 2500 ms  user leaves the detail page

shareReplay(1)'in anlamı olan refCount: false ile akış, son subscriber ayrıldıktan sonra da üretmeye devam etti: 10, 9, 8 ve uygulama çalıştıkça devam ederdi (örnek sanal saati 5. saniyede durduruyor). ReplaySubject kaynağa olan subscription'ını tutuyor, kaynak timer'ını ya da socket'ini tutuyor ve pipe'ın yakaladığı her şey bellekte kalıyor. refCount: true ile akış son subscriber'la birlikte durdu. Kural: hiç tamamlanmayan bir kaynak için (interval, WebSocket, bir store selector'ı, valueChanges) shareReplay({ bufferSize: 1, refCount: true }) yaz. refCount: false yalnız kaynak tamamlanıyorsa ya da bilerek bütün oturum boyunca çalışmasını istiyorsan doğru.

Başarısız Yanıtı Cache'lemek

Cache'lenen bir isteğin ikinci tuzağı, istek başarısız olduğunda neyin cache'lendiği. Burada kategori endpoint'ine yapılan ilk çağrı 503 ile başarısız oluyor, sonraki bütün çağrılar başarılı:

import { catchError, defer, Observable, of, shareReplay, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

// ilk çağrı 503 ile başarısız, sonrasında sunucu sağlıklı
function categoriesApi(): Observable<string[]> {
  let calls = 0;
  return defer(() =>
    ++calls === 1
      ? new Observable<string[]>((subscriber) => {
          log("-> GET /categories");
          const pending = timer(200).subscribe(() => {
            log("<- GET /categories 503");
            subscriber.error(new Error("503 Service Unavailable"));
          });
          return () => pending.unsubscribe();
        })
      : fakeRequest("GET /categories", 200, () => ["science-fiction", "classics", "fantasy"]),
  );
}

function render(categories$: Observable<string[]>, label: string) {
  categories$.subscribe({
    next: (c) => log(`${label}: [${c.join(", ")}]`),
    error: (err: Error) => log(`${label}: error ${err.message}`),
  });
}

runInVirtualTime("catchError before shareReplay: the fallback is cached", () => {
  const categories$ = categoriesApi().pipe(
    catchError(() => of([])),
    shareReplay(1),
  );
  render(categories$, "menu");
  timer(5000).subscribe(() => render(categories$, "menu after 5 s"));
});

runInVirtualTime("shareReplay alone: the error is not cached", () => {
  const categories$ = categoriesApi().pipe(shareReplay(1));
  render(categories$.pipe(catchError(() => of([]))), "menu");
  timer(5000).subscribe(() => render(categories$.pipe(catchError(() => of([]))), "menu after 5 s"));
});

--- catchError before shareReplay: the fallback is cached
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  200 ms  menu: []
 5000 ms  menu after 5 s: []
--- shareReplay alone: the error is not cached
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  200 ms  menu: []
 5000 ms  -> GET /categories
 5200 ms  <- GET /categories
 5200 ms  menu after 5 s: [science-fiction, classics, fantasy]

İlk sürümde catchError, shareReplay'in önünde duruyor ve hatayı boş bir listeye çeviriyor. shareReplay için bu boş liste, ardından tamamlanma gelen başarılı bir cevap; dolayısıyla cache'leniyor. Beş saniye sonra sunucu yeniden sağlıklıydı, ama menü istek bile atmadan boş liste gösterdi ve sayfa yenilenene kadar boş kalacaktı. İkinci sürümde hata shareReplay'e ulaşıyor, o da hatada reset oluyor (resetOnError: true) ve her subscriber hatayı kendisi ele alıyor. 5. saniyedeki subscriber yeni bir istek attı ve kategorileri aldı. Yedek değeri paylaşımdan sonraya, tüketen tarafa koy; paylaşılan kısım ya başarılı olmalı ya da hata vermeli, yerine bir şey saklamamalı. catchError ve retry'ın ayrıntıları bu serinin error handling yazısında.

connectable ve connect: Kontrollü Multicasting

share, kaynağı ilk subscriber geldiğinde başlatıyor. Bazen anı kendin belirlemek istersin, örneğin ilk değer üretilmeden önce bütün subscriber'ların yerinde olduğundan emin olmak için. connectable bunu yapıyor; connect operator'ü ise benzer bir sorunu tek bir pipe içinde çözüyor:

import { connect, connectable, defer, filter, interval, map, merge, take, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// Dune'un stok akışı: her bağlantıyı loglar, birer saniye arayla 12, 11, 10 yayar
const stockFeed = defer(() => {
  log("feed connected");
  return interval(1000).pipe(
    map((i) => 12 - i),
    take(3),
  );
});

runInVirtualTime("connectable: subscribe first, connect when ready", () => {
  const stock$ = connectable(stockFeed);
  stock$.subscribe((s) => log(`header badge: ${s}`));
  stock$.subscribe((s) => log(`detail page: ${s}`));
  timer(500).subscribe(() => {
    log("connect()");
    stock$.connect(); // iki subscriber için akışa tek subscription
  });
});

runInVirtualTime("connect operator: share inside one pipe", () => {
  stockFeed
    .pipe(
      connect((shared$) =>
        merge(
          shared$.pipe(map((s) => `stock: ${s}`)),
          shared$.pipe(
            filter((s) => s <= 10),
            map((s) => `low stock warning: ${s}`),
          ),
        ),
      ),
    )
    .subscribe((line) => log(line));
});

--- connectable: subscribe first, connect when ready
  500 ms  connect()
  500 ms  feed connected
 1500 ms  header badge: 12
 1500 ms  detail page: 12
 2500 ms  header badge: 11
 2500 ms  detail page: 11
 3500 ms  header badge: 10
 3500 ms  detail page: 10
--- connect operator: share inside one pipe
    0 ms  feed connected
 1000 ms  stock: 12
 2000 ms  stock: 11
 3000 ms  stock: 10
 3000 ms  low stock warning: 10

connectable(source), subscribe edildiğinde hiçbir şey yapmayan bir Observable döndürüyor; header rozeti ve detay sayfası ona bağlandı, akışa ise 500. ms'deki connect() ikisi için bir kez subscribe oldu. Seçenekleri connector (varsayılanı Subject) ve resetOnDisconnect. connect operator'ü ise bir fonksiyona kaynağın paylaşılan bir sürümünü veriyor: merge içindeki iki kol aynı akışı okuyor ve logda tek bir "feed connected" var. connect olmadan stockFeed üzerindeki iki pipe iki bağlantı demek olurdu.

Eski bir kodu sürdürüyorsan bunların yerine publish, publishReplay, multicast ve refCount() görürsün. RxJS 7'de bunlar deprecated; 7.8.2'nin type tanımları 8. sürümde kaldırılacaklarını yazıyor ve connectable, connect ve share'e yönlendiriyor. Örneğin publishReplay(1), refCount(), shareReplay({ bufferSize: 1, refCount: true }) oluyor. Geçişi bu serinin RxJS 9 yazısı anlatıyor.

Üretimde Dikkat

Varsayımı değil isteği say. Tarayıcının Network sekmesinde /api/ filtresi ya da API'nin access logu tekrarlanan istekleri hemen gösteriyor; aynı GET'i birkaç milisaniye arayla iki kez gönderen bir sayfada neredeyse her zaman cold bir Observable'a iki subscription vardır.

Uygulama çapındaki bir shareReplay bir cache'tir ve her cache "taze veriyi nasıl alır?" sorusuna bir cevap ister. Kategoriler bir oturum boyunca yaşayabilir; fiyatlar ve stok yaşayamaz. Yukarıdaki gibi süreli reset'li bir share ayarı kullan ya da bir yazma işleminden sonra (ör. bir admin kategori eklediğinde) paylaşılan Observable'ı yeniden oluşturarak cache'i açıkça geçersiz kıl. Angular'da cache tutan bir service kullanıcı çıkış yaptığında da temizlenmeli; yoksa aynı tarayıcıdaki bir sonraki kullanıcı öncekinin verisini görür.

Son olarak, cache'teki değer referansla paylaşılıyor. Bir component aldığı array'i değiştirirse (push, sort), diğer bütün subscriber'lar da değişikliği görür. Paylaşılan değerleri immutable kabul et: sıralamadan önce kopyala (toSorted) ya da geliştirme ortamında dondur.

Sık Düşülen Tuzaklar

• Aynı HttpClient Observable'ını iki async pipe ile ya da async ve subscribe ile kullanmak. Her biri kendi isteğini gönderir.
• share()'in cache'lemesini beklemek. Kaynak tamamlanınca reset olur; ondan sonra gelen subscriber kaynağı yeniden başlatır. shareReplay ya da ReplaySubject connector'ı kullan.
• shareReplay'i bir method içinde çağırmak. Her çağrı yeni bir paylaşılan Observable oluşturur; bir field'da sakla.
• Hiç tamamlanmayan bir kaynakta shareReplay(1) kullanmak. refCount: false ile kaynak sonsuza kadar çalışır; refCount: true yaz.
• catchError'ı shareReplay'in önüne koymak. Yedek değer cache'lenir ve hata hiç geçmez; hatayı paylaşımdan sonra ele al.
• Paylaşılan, cache'lenmiş bir değeri değiştirmek. Bütün subscriber'lar aynı referansı tutuyor.
• Yeni kodu publish, multicast ya da refCount() ile yazmak. RxJS 7'de deprecated.

Küçük Bir Deneme

★☆☆ 01-predict.ts'te share() yerine shareReplay(1) koy. Sunucu logu şimdi kaç istek gösteriyor? Sonra share({ resetOnComplete: false }) ile footer'ın subscription'ını 0 ms'ye alıp 1000 ms'ye geri koy; shareReplay(1)'e göre ne değişti, ne değişmedi?

★★☆ CategoryService'e, bir sonraki subscriber'ın listeyi yeniden çekmesini sağlayan bir refresh() method'u ekle (ör. bir admin kategori ekledikten sonra); zaten subscribe olmuş component'leri bozmasın. İpucu: paylaşılan Observable'ı bir field'da tutup değiştir ya da share için reset bildiricisi olarak bir Subject kullan.

★★★ 03-sharereplay-refcount.ts'i, kullanıcı detay sayfasından 2,5. saniyede çıkıp 4. saniyede geri dönecek şekilde değiştir. refCount: true ile sayfa döndükten hemen sonra ne gösteriyor, akış devam mı ediyor yoksa baştan mı başlıyor? Sonra share({ connector: () => new ReplaySubject(1), resetOnRefCountZero: () => timer(3000) })'i dene ve farkı açıkla.

Bu yazıdaki iki senaryo bir hatayla bitti: başarısız kategori isteği ve yanlışlıkla cache'lenen 503. Bir sonraki yazıda hataları doğru şekilde ele alacağız: catchError'ın gerçekte neyi değiştirdiğini, başarısız bir istekten sonra arama kutusunu nasıl yaşatacağımızı, retry'ı exponential backoff ile yalnız güvenli olduğu yerde, Angular'ın HttpErrorResponse'unu ve ASP.NET Core'un ProblemDetails'ini ve loading göstergeleri için finalize'ı göreceğiz.