RxJS Subject Türleri: Subject, BehaviorSubject, ReplaySubject, AsyncSubject ve Ne Zaman Hiçbiri

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'da (bir online kitapçı) Dune'un stoku birer birer azalıyor: 12, 11, 10, 9. Bu sayıları dört farklı Subject türüne gönderiyor ve bir subscriber'ı geç, 10'dan sonra katılıyoruz:

import { AsyncSubject, BehaviorSubject, ReplaySubject, Subject } from "rxjs";

// Dune'un stoku 12 -> 11 -> 10 -> 9; geç gelen subscriber 10'dan sonra katılıyor
const subjects = {
  Subject: new Subject<number>(),
  BehaviorSubject: new BehaviorSubject<number>(12),
  "ReplaySubject(2)": new ReplaySubject<number>(2),
  AsyncSubject: new AsyncSubject<number>(),
};

for (const [name, subject] of Object.entries(subjects)) {
  const seen: string[] = [];
  subject.next(11);
  subject.next(10);
  subject.subscribe({ next: (stock) => seen.push(String(stock)), complete: () => seen.push("|") });
  subject.next(9);
  subject.complete();
  console.log(`${name.padEnd(17)} late subscriber saw: ${seen.join(" ")}`);
}

Okumaya devam etmeden önce dört satırın her birinde geç gelen subscriber'ın ne gördüğünü yaz. Hangisi 12'yi görür?

Şimdiye kadar kurduğumuz her Observable'ın kendi producer'ı vardı: bir timer, bir event, bir HTTP isteği. Subject farklı: next'i kendin çağırıyorsun ve değer bütün subscriber'lara gidiyor. Angular serisinin "Angular'da RxJS'e Giriş" yazısında bir service içinde BehaviorSubject kullanmıştık; bu yazı bir adım ileri gidiyor. Dört Subject türünü karşılaştıracak, geç gelen subscriber'a ne verdiklerine bakacak, ölü bir Subject'in neden sessizce başarısız olduğunu ve public bir field'daki Subject'in neden sorun çıkardığını görecek, sonra Angular 22'de neyin Subject'e, neyin signal'a ait olduğuna karar vereceğiz.

Subject Hem Observable Hem Observer

Cevap:

Subject           late subscriber saw: 9 |
BehaviorSubject   late subscriber saw: 10 9 |
ReplaySubject(2)  late subscriber saw: 11 10 9 |
AsyncSubject      late subscriber saw: 9 |

Subject'in bir Observable gibi subscribe'ı, bir Observer gibi next, error ve complete'i var. Subscriber'larının listesini tutuyor ve her bildirimi hepsine iletiyor; yani doğası gereği hot: değerler, dinleyen olsun olmasın var. Dört tür yalnız tek bir şeyde ayrılıyor: subscriber'ın subscribe olduğu anda ne aldığı.

Çıktıda Subject yalnız 9'u verdi, yani subscribe'dan sonra gelen değeri. BehaviorSubject hemen 10'u verdi, çünkü her zaman güncel değeri tutuyor; 12 başlangıç değeriydi ve çoktan değişmişti. ReplaySubject(2) son iki değeri (11, 10) tekrar oynattı, sonra 9 geldi. AsyncSubject 9'u ancak complete çağrılınca verdi; Subject hiç tamamlanmasaydı subscriber hiçbir şey görmezdi. AsyncSubject'in uygulama kodunda nadir olmasının sebebi bu: bir Promise gibi davranıyor ve firstValueFrom, lastValueFrom ya da bir sonraki yazının multicasting operator'leri aynı ihtiyacı daha açık karşılıyor.

ReplaySubject: Bir Zaman Penceresinde Son N Değer

ReplaySubject iki sınır alıyor: kaç değer tutulacağı ve ne kadar süre. Kitap detay sayfası "son bakılan" kitapları gösteriyor; en fazla üç tane ve yalnız son on saniyedekileri istiyoruz:

import { ReplaySubject, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

runInVirtualTime("ReplaySubject(3, 10000): recently viewed books", () => {
  // en fazla 3 görüntüleme ve yalnız son 10 saniyedekiler
  const viewed = new ReplaySubject<string>(3, 10_000);
  const views: [number, string][] = [[0, "Dune"], [2000, "Emma"], [4000, "The Hobbit"], [6000, "Middlemarch"], [13000, "Frankenstein"]];
  for (const [ms, title] of views) timer(ms).subscribe(() => viewed.next(title));

  for (const ms of [7000, 15000]) {
    timer(ms).subscribe(() => {
      const replayed: string[] = [];
      viewed.subscribe((title) => replayed.push(title)).unsubscribe();
      log(`late subscriber gets: ${replayed.join(", ")}`);
    });
  }
});

--- ReplaySubject(3, 10000): recently viewed books
 7000 ms  late subscriber gets: Emma, The Hobbit, Middlemarch
15000 ms  late subscriber gets: Middlemarch, Frankenstein

7. saniyede dört kitaba bakılmıştı, buffer son üçünü tuttu. 15. saniyede işi pencere yaptı: Emma (2. sn) ve The Hobbit (4. sn) on saniyeden eskiydi, geriye yalnız Middlemarch ve Frankenstein kaldı. Pencere, değer eklendiğinde ve biri subscribe olduğunda kontrol ediliyor; eski değerler bellekte sonsuza kadar kalmıyor. Argümansız ReplaySubject() ise her şeyi tutuyor; uzun yaşayan bir akışta bu bir memory leak, bu yüzden en azından buffer boyutunu her zaman ver.

ReplaySubject(1) ile BehaviorSubject birbirine benziyor ama aynı değil. BehaviorSubject bir başlangıç değeri istiyor ve hiç değer gönderilmemiş olsa bile ilk subscriber'a onu veriyor; ReplaySubject(1) ilk next'e kadar hiçbir şey vermiyor. "Henüz değer yok" gerçek bir durumsa (kullanıcı giriş yapmadı, veri gelmedi), ReplaySubject(1) bunu null ile başlatılmış bir BehaviorSubject'ten daha doğru anlatıyor.

complete ve error Sonrası: Ölü Subject

Subject de her Observable'la aynı contract'a uyuyor: complete ya da error'dan sonra kalıcı olarak bitiyor. Bunu tehlikeli yapan, sessizce başarısız olması:

import { BehaviorSubject, Subject } from "rxjs";

const cartEvents = new Subject<string>();
cartEvents.subscribe({ next: (e) => console.log("A next:", e), complete: () => console.log("A complete") });
cartEvents.next("Dune added");
cartEvents.complete();
cartEvents.next("Emma added"); // yok sayılır, hata da yok
cartEvents.subscribe({ next: (e) => console.log("B next:", e), complete: () => console.log("B complete") });

const stock = new BehaviorSubject(12);
stock.complete();
stock.subscribe({ next: (n) => console.log("C next:", n), complete: () => console.log("C complete") });
console.log("getValue() after complete:", stock.getValue());

const prices = new Subject<number>();
prices.error(new Error("price feed lost"));
prices.subscribe({ next: (p) => console.log("D next:", p), error: (err: Error) => console.log("D error:", err.message) });

A next: Dune added
A complete
B complete
C complete
getValue() after complete: 12
D error: price feed lost

"Emma added" kimseye ulaşmadı ve hata da çıkmadı; durmuş bir Subject'te next hiçbir şey yapmıyor. complete'ten sonra gelen B yalnız tamamlanmayı aldı. BehaviorSubject güncel değerini C'ye bile vermedi, oysa getValue() hâlâ 12 döndürüyor. error'dan sonra ise her yeni subscriber aynı hatayı hemen alıyor. Pratikte bu, birisi service seviyesindeki bir Subject'te bir component'in ngOnDestroy'unda complete() çağırdığında oluyor: service yaşamaya devam ediyor, ama Subject'i sonradan gelen her component için ölü. Bir Subject'i yalnız sahibi tamamlamalı; uygulama boyunca yaşayan bir Subject'in ise genellikle hiç tamamlanmasına gerek yok.

asObservable ile Dışarıya Salt Okunur Görünüm Vermek

Public bir field'daki Subject, onu görebilen herkesin değer yaymasına izin veriyor. Subject'i açmak ile ona salt okunur bir görünüm açmak arasındaki fark:

import { Subject, type Observable } from "rxjs";

class LeakyCartEvents {
  readonly added$ = new Subject<string>(); // herkes next çağırabilir
}

class CartEvents {
  readonly #added = new Subject<string>();
  readonly added$: Observable<string> = this.#added.asObservable(); // salt okunur görünüm

  notifyAdded(title: string): void {
    this.#added.next(title);
  }
}

const leaky = new LeakyCartEvents();
leaky.added$.subscribe((title) => console.log(`toast: ${title} added to cart`));
leaky.added$.next("Free Bookmark"); // uzaktaki bir component "yardım ediyor" ve sahte bir event yayıyor

const events = new CartEvents();
events.added$.subscribe((title) => console.log(`toast: ${title} added to cart`));
events.notifyAdded("Dune");
console.log("added$ has next:", "next" in events.added$);

toast: Free Bookmark added to cart
toast: Dune added to cart
added$ has next: false

İlk class'ta uzaktaki bir component added$ üzerinde next çağırdı ve toast kimsenin eklemediği bir ayraç hediyesini duyurdu. Kodda bunu kimin yaptığını gösteren hiçbir şey yok; projede .next( araman gerekiyor. İkinci class'ta Subject private bir field (#added) ve dışarısı yalnız asObservable() alıyor, yani next'i olmayan bir Observable. Değer yaymanın tek yolu notifyAdded method'u; onu bulabilir, test edebilir ve loglayabilirsin. Aynı kural signal'lar için de geçerli: Angular'ın asReadonly()'si değeri set olmadan dışarı veriyor, birazdan göreceğiz.

Depo Örneğine Dönüş: Event mi, State mi?

Creation function'lar yazısındaki ★★☆ alıştırma, B subscriber'ının kaçırdığı depo event'lerini nasıl görebileceğini ve bunun neyi bozacağını sormuştu. Subject'lerle cevabı:

import { ReplaySubject, Subject, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// creation function'lar yazısındaki depo örneği, bu kez Subject'lerle
runInVirtualTime("Subject: B misses earlier events", () => {
  const restocked = new Subject<string>();
  timer(100).subscribe(() => restocked.next("Dune"));
  timer(300).subscribe(() => restocked.next("Emma"));
  restocked.subscribe((title) => log(`A: ${title} restocked`));
  timer(250).subscribe(() => restocked.subscribe((title) => log(`B: ${title} restocked`)));
});

runInVirtualTime("ReplaySubject(1): B gets the last event, even when it is stale", () => {
  const restocked = new ReplaySubject<string>(1);
  const soldOut = new Subject<string>();
  timer(100).subscribe(() => restocked.next("Dune"));
  timer(200).subscribe(() => soldOut.next("Dune")); // Dune yeniden tükeniyor, restocked$'ın bundan haberi yok
  soldOut.subscribe((title) => log(`warehouse: ${title} sold out`));
  timer(250).subscribe(() => restocked.subscribe((title) => log(`B: ${title} restocked (replayed)`)));
});

--- Subject: B misses earlier events
  100 ms  A: Dune restocked
  300 ms  A: Emma restocked
  300 ms  B: Emma restocked
--- ReplaySubject(1): B gets the last event, even when it is stale
  200 ms  warehouse: Dune sold out
  250 ms  B: Dune restocked (replayed)

Düz Subject'te B 250. ms'de katıldı ve "Dune restocked"ı kaçırdı. ReplaySubject(1) bunu çözüyor, ama ikinci senaryo alıştırmanın ima ettiği sorunu gösteriyor: Dune 200. ms'de yeniden tükendi ve 250. ms'de katılan B'ye hâlâ Dune'un az önce stoğa girdiği söylendi. Tekrar oynatılan mesaj olduğu anda doğruydu, şu an yanlış.

Bu yazının en işe yarar dersi bu. Event "bir şey oldu" der; state "durum şu" der. Bir event'i sonradan katılana tekrar oynatmak, geçmişi yanlış bir şimdiye çevirir. B, Dune'un stokta olup olmadığını bilmek istiyorsa state'e ihtiyacı var: hem "restocked" hem "sold out" event'lerinin güncellediği, kitap başına bir stok değeri. Eski Angular kodunda BehaviorSubject'in kullanıldığı yer tam burasıydı; Angular 22'de burada signal kullanılıyor.

Angular 22: State Signal'da, Event Subject'te

Bookstore'un Angular istemcisi bookstore-web'in artık bir sepeti var. Snapshot'ta proje bookstore-web/ klasöründe (orada npm install ve npm start; repo kökünde node api/server.ts çalışırken). Sepette iki tür bilgi var: sepette şu anda ne olduğu (header'daki rozet ve sepet sayfası bunu her an gösteriyor) ve bir kitabın az önce eklendiği (toast bunu bir kez gösteriyor). İlki state, ikincisi event:

import { computed, Service, signal } from '@angular/core';
import { Subject, type Observable } from 'rxjs';
import type { Book } from '../catalog/book';

export interface CartItem {
  book: Book;
  quantity: number;
}

@Service()
export class CartStore {
  // state: sepette şu anda ne var -> signal
  readonly #items = signal<CartItem[]>([]);
  readonly items = this.#items.asReadonly();
  readonly count = computed(() => this.#items().reduce((sum, i) => sum + i.quantity, 0));
  readonly total = computed(() => this.#items().reduce((sum, i) => sum + i.quantity * i.book.price, 0));

  // event: bir şey az önce oldu -> Subject, dışarıya salt okunur
  readonly #added = new Subject<Book>();
  readonly added$: Observable<Book> = this.#added.asObservable();

  add(book: Book): void {
    this.#items.update((items) => {
      const existing = items.find((i) => i.book.isbn === book.isbn);
      return existing
        ? items.map((i) => (i === existing ? { ...i, quantity: i.quantity + 1 } : i))
        : [...items, { book, quantity: 1 }];
    });
    this.#added.next(book);
  }
}

@Service(), Angular 22'de ng generate service'in ürettiği decorator; @Injectable({ providedIn: 'root' })'un kısa hâli. Ürünler private, yazılabilir bir signal'da duruyor ve dışarıya asReadonly() ile çıkıyor; count ve total computed, yani header yalnız cart.count()'u okuyor ve signal değişince güncelleniyor. "Eklendi" event'i ise asObservable() görünümü olan private bir Subject. Toast component'i bunu üç saniye sonra kaybolan bir mesaja çeviriyor:

import { Component, inject } from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { concat, map, of, switchMap, timer } from 'rxjs';
import { CartStore } from './cart-store';

@Component({
  selector: 'app-cart-toast',
  template: `
    @if (message(); as text) {
      <p class="toast" role="status">{{ text }}</p>
    }
  `,
})
export class CartToast {
  readonly message = toSignal(
    inject(CartStore).added$.pipe(
      // mesajı 3 saniye göster; yeni bir event mevcut mesajın yerini alır
      switchMap((book) =>
        concat(of(`${book.title} added to cart`), timer(3000).pipe(map(() => null))),
      ),
    ),
    { initialValue: null },
  );
}

Bu, signal'ın iyi yapamadığı, RxJS'in ise tek satırda yaptığı bir iş: her event üç saniyelik bir timer başlatıyor, yeni bir event switchMap üzerinden eski timer'ı iptal ediyor. toSignal sonucu template için bir signal'a çeviriyor ve component destroy olunca unsubscribe oluyor; ayrıntıları bu serinin Angular yazısında. Katalog sayfası buton tıklamasında cart.add(book) çağırıyor. Headless Chromium'da önce Dune'a, bir saniye sonra Emma'ya tıkladık:

 0.4 s catalog loaded         | Cart (0): 0 TRY | (no toast)
 0.5 s click Dune             | Cart (1): 450 TRY | Dune added to cart
 1.7 s click Emma 1 s later   | Cart (2): 770 TRY | Emma added to cart
 4.2 s 2.5 s later            | Cart (2): 770 TRY | Emma added to cart
 5.2 s another 1 s later      | Cart (2): 770 TRY | (no toast)
Frankenstein button disabled: true

Header her tıklamada signal'ı takip etti. Toast 1,7. saniyede Emma'ya geçti ve 4,2. saniyede hâlâ görünüyordu: Dune'un timer'ı yaklaşık 3,5. saniyede bitecekti, ama switchMap onu iptal edip Emma için yeni bir timer başlatmıştı; o da yaklaşık 4,7. saniyede bitti. Frankenstein'ın butonu devre dışı, çünkü stoku 0.

Peki sepet neden "Angular'da RxJS'e Giriş" yazısındaki gibi bir BehaviorSubject'te durmuyor? Hâlâ çalışır, ama Angular 22'de change detection zoneless ve component'ler varsayılan olarak OnPush (Angular serisinin "Angular 22: Varsayılan OnPush, Stable Signal Forms ve Angular 12 ile Proje Farkı" yazısı). BehaviorSubject, onu okuyan her template'te bir async pipe ya da toSignal istiyor; güncel değere getValue() ya da bir subscription ile ulaşılıyor; türetilmiş değerler için map artı combineLatest gerekiyor. Signal ise senkron okunuyor, computed ile türetiliyor ve Angular'a tam olarak hangi view'ın güncelleneceğini söylüyor. Tersinden bakınca da signal kötü bir event kanalı: aynı değeri iki kez set etmek değişiklik sayılmıyor ve hiçbir şey "bir sonrakini" beklemiyor. İş bölümü basit: state signal'a, event'ler Subject'e (ya da başka bir kaynaktan gelen Observable'a), zamanla ilgili işler RxJS'e.

Ne Zaman Hiçbiri: Global Event Bus

Subject'in popüler kullanımlarından biri uygulama çapında bir event bus: tek bir Subject, string event tipleri ve uygulamanın her parçası hem yayıyor hem dinliyor:

import { filter, map, Subject } from "rxjs";

// uygulama çapında bir "event bus": tek Subject, string event tipleri, her şekilde payload
type BusEvent = { type: string; payload: unknown };
const bus = new Subject<BusEvent>();
const on = (type: string) =>
  bus.pipe(
    filter((e) => e.type === type),
    map((e) => e.payload),
  );

on("cart:added").subscribe((p) => console.log("header badge got", p));
on("cart:add").subscribe((p) => console.log("toast got", p)); // yazım hatası: derleniyor, hiç tetiklenmiyor

bus.next({ type: "cart:added", payload: { isbn: "9780441172719", title: "Dune" } });
bus.next({ type: "cart:added", payload: "Emma" }); // farklı bir payload şekli, bu da derleniyor

header badge got { isbn: '9780441172719', title: 'Dune' }
header badge got Emma

Toast hiç tetiklenmedi, çünkü listener'ı cart:added yerine cart:add'i bekliyor; derleyici yardım edemiyor, çünkü ikisi de yalnızca string. Header rozeti iki farklı payload şekli aldı ve kendini ikisine karşı da korumak zorunda. Her hata avında sorulan "bu event'i kim gönderiyor?" sorusunun ise tek bir cevabı var: string'i bütün kodda aramak. Bus kodu o kadar iyi ayırıyor ki artık kimse akışı takip edemiyor.

Bir Subject oluşturmadan önce ona gerçekten ihtiyaç olup olmadığını sor. Değer başka bir yerden geliyorsa (DOM event'i, HTTP çağrısı, router event'i), fromEvent ya da API'nin kendi Observable'ıyla doğrudan onu kullan. State ise signal kullan. Bir child component parent'ına haber veriyorsa output() kullan. Subject, kendi kodunun ürettiği ve birbirinden bağımsız birkaç parçanın tükettiği event'ler için doğru araç; o zaman bile global bir bus'ta değil, CartStore gibi tipli bir service'in içinde durmalı.

Üretimde Dikkat

Root service'lerdeki Subject'ler uygulama kadar yaşıyor. Subscriber'ları genellikle öyle değil: elle subscribe olup unsubscribe olmayı unutan her component, bütün view'ıyla bellekte kalıyor. Destroy anında temizleyen toSignal'ı, async pipe'ı ya da takeUntilDestroyed'ı tercih et. Chrome DevTools'un Memory panelinde, sayfalar arasında birkaç kez gidip geldikten sonra alınan bir heap snapshot bu tür sızıntıları sayısı sürekli artan component instance'ları ve detached DOM element'leri olarak gösteriyor.

Aynı Subject'in bir subscriber'ı içinden next çağırmak da sürpriz kaynağı: içteki değer kalan subscriber'lara dıştakinden önce ulaşıyor, yani onlar değerleri farklı sırayla görebiliyor. Bir event başka bir event'i tetikleyecekse bunu subscribe içinde next ile değil, bir pipe ile (map, switchMap) modelle.

Son olarak, logu sahibinde tut. Her yayın notifyAdded gibi bir method'dan geçtiği için oraya eklenen tek bir satır (ya da private Subject üzerinde bir tap) her event'i payload'ıyla kaydediyor; global bir bus'ın imkânsız kıldığı şey tam olarak bu.

Sık Düşülen Tuzaklar

• Subject'i public bir field olarak açmak. Herkes next çağırabilir; private tut ve asObservable()'ı aç.
• Paylaşılan bir Subject'te component'in ngOnDestroy'unda complete() çağırmak. Sonradan gelen her subscriber yalnız tamamlanmayı alır, sonraki next çağrıları hata vermeden yok sayılır.
• Event'leri ReplaySubject ile geç gelen subscriber'lara tekrar oynatmak. Tekrar oynatılan bir event artık var olmayan bir durumu anlatabilir; state'i state olarak tut.
• Uzun yaşayan bir akışta buffer boyutu vermeden ReplaySubject() kullanmak. Her değeri sonsuza kadar tutar.
• "Henüz yüklenmedi" demek için BehaviorSubject'i null ile başlatıp her yerde null süzmek. ReplaySubject(1) ya da açık bir loading durumu olan bir signal bunu daha iyi anlatır.
• String event tipleriyle global bir event bus kurmak. Yazım hataları derlenir, payload'ların tipi yoktur ve göndereni kimse bulamaz.

Küçük Bir Deneme

★☆☆ 01-predict.ts'te geç gelen subscribe'ı complete()'ten sonraya taşı. Çalıştırmadan önce her satırı tahmin et; hangi Subject geç gelen subscriber'a hâlâ bir değer veriyor?

★★☆ Depo örneğini düzelt: başlık başına stoku tutan bir BehaviorSubject<Record<string, number>> kur, onu hem "restocked" hem "sold out" ile güncelle ve B'nin "Dune stokta mı?" bilgisini map ve distinctUntilChanged ile türetmesini sağla. Sonra aynısını bookstore-web'de bir signal ve computed ile yaz ve kod miktarını karşılaştır.

★★★ Sepet toast'ına bir "Undo" butonu ekle: tıklama az önce eklenen kitabı çıkarsın ve toast'ı hemen gizlesin. CartStore'un Subject'ini private tut ve "undo"nun yeni bir event mi, bir state değişikliği mi, yoksa ikisi birden mi olduğuna karar ver.

Subject, kendi kodumuzun ürettiği değerleri paylaşıyor. Bir sonraki yazıda ise var olan bir Observable'ı paylaşacağız: aynı HttpClient çağrısına bağlanan iki async pipe'ın neden iki istek gönderdiğini ve share, shareReplay ve connectable'ın cold bir Observable'ı, belleği sızdırmadan ve başarısız bir yanıtı sonsuza kadar cache'lemeden nasıl paylaşılan bir Observable'a çevirdiğini göreceğiz.