RxJS'te Observable'ı İçinden Anlamak: Contract, Lazy Çalışma, Teardown ve 40 Satırlık Kendi Observable'ın

Sürüm notu: Bu yazıdaki örnekler RxJS 7.8.2, TypeScript 6.0 ve Node.js 24 ile denenmiştir; tarayıcı davranışı Playwright 1.65'in Chromium build'inde gözlenmiştir.

Önce Tahmin Et

Aşağıdaki kod sitedeki serilerin ortak örnek uygulaması Bookstore'dan (bir online kitapçı): Dune'un stok bilgisini yayan bir Observable. Çalıştırmadan önce konsolda hangi satırların hangi sırayla çıkacağını kâğıda yaz.

import { Observable } from "rxjs";

const duneStock$ = new Observable<number>((subscriber) => {
  console.log("producer started");
  subscriber.next(12);
  subscriber.next(11);
  subscriber.complete();
  subscriber.next(10);
});

console.log("before subscribe");
duneStock$.subscribe((stock) => console.log("A got", stock));
duneStock$.subscribe((stock) => console.log("B got", stock));
console.log("after subscribe");

Üç soru: "producer started" kaç kez yazılır? 10 değeri ekrana gelir mi? "after subscribe" satırı stok değerlerinden önce mi çıkar, sonra mı? Çoğu kişi Observable'ı "asenkron bir şey" diye öğrendiği için "after subscribe"ın önce geleceğini söyler. Cevap biraz aşağıda.

Angular serisinin "Angular'da RxJS'e Giriş" yazısında Observable ile Promise farkını, ilk operator'leri, BehaviorSubject'li service'i ve async pipe'ı gördük. Bu seri oradan devam ediyor: amaç "hangi operator, neden" sorusuna gerekçeli cevap verebilmek. İlk yazıda Observable'ın içine bakıyoruz: contract'ı, lazy çalışması, teardown ve sonunda RxJS'in sihir olmadığını göstermek için 40 satırlık kendi Observable'ımız.

Örnekleri Çalıştırmak

Bu yazının kodu serinin snapshot'ında src/01/ klasöründe. Proje bir Angular uygulaması değil, düz bir Node.js projesi; RxJS'in kendisini Angular'dan bağımsız görmek daha kolay. Node.js 24, .ts dosyalarını type'ları silerek doğrudan çalıştırabildiği için build adımı yok:

npm install
node src/01/01-predict.ts

before subscribe
producer started
A got 12
A got 11
producer started
B got 12
B got 11
after subscribe

Tahminin cevabı burada. "producer started" iki kez yazıldı, çünkü her subscribe çağrısı new Observable'a verdiğin fonksiyonu baştan çalıştırıyor. 10 hiç gelmedi, çünkü complete()'ten sonra gönderilen değer contract'a aykırı. "after subscribe" en sonda, çünkü bu producer'da asenkron hiçbir şey yok: değerler subscribe çağrısının içinde, senkron olarak teslim edildi. Observable asenkron olmak zorunda değil; asenkron olup olmadığına producer karar veriyor.

Push ve Pull: Observable Tablonun Neresinde?

JavaScript'te bir değeri üreten taraf (producer) ile kullanan taraf (consumer) arasında dört temel ilişki var. Pull'da consumer "şimdi ver" der; push'ta producer "hazır olunca gönderirim" der. Bir de değer sayısı var: bir tane mi, birden çok mu?

Dördünü aynı veriyle yan yana koyalım:

import { Observable } from "rxjs";

// 1) Function: pull, tek değer
function readStock(): number {
  return 12;
}
console.log("function:", readStock());

// 2) Iterator: pull, çok değer; next()'i consumer çağırır
function* stockHistory() {
  yield 12;
  yield 11;
  yield 10;
}
const it = stockHistory();
console.log("iterator:", it.next().value, it.next().value);

// 3) Promise: push, tek değer; hemen çalışmaya başlar
const stockPromise = new Promise<number>((resolve) => {
  console.log("promise executor runs now");
  setTimeout(() => resolve(12), 10);
});

// 4) Observable: push, çok değer; subscribe olmadan hiçbir şey çalışmaz
const stock$ = new Observable<number>((subscriber) => {
  console.log("observable producer runs now");
  subscriber.next(12);
  setTimeout(() => {
    subscriber.next(11);
    subscriber.complete();
  }, 20);
});

console.log("--- subscribing ---");
stockPromise.then((v) => console.log("promise:", v));
stock$.subscribe({
  next: (v) => console.log("observable:", v),
  complete: () => console.log("observable: complete"),
});

function: 12
iterator: 12 11
promise executor runs now
--- subscribing ---
observable producer runs now
observable: 12
promise: 12
observable: 11
observable: complete

İki satıra dikkat et. "promise executor runs now", "--- subscribing ---" satırından önce çıktı: Promise, kimse then çağırmadan işe başladı. Observable'ın producer'ı ise ancak subscribe ile çalıştı ve ilk değeri (12) senkron, ikincisini 20 ms sonra asenkron gönderdi. Aynı Observable hem senkron hem asenkron değer üretebiliyor; Promise'te bu mümkün değil, then callback'i her zaman bir microtask sonra çalışır (bu sıralamanın nedeni JavaScript İleri serisinin "JavaScript Event Loop Derinlemesine" yazısında).

Observable'ı en iyi anlatan benzetme şu: Observable, değer döndürmek yerine değerleri bir callback'e gönderen bir fonksiyondur. subscribe etmek o fonksiyonu çağırmaktır. Fonksiyonu iki kez çağırırsan gövdesi iki kez çalışır; Observable'a iki kez subscribe olursan producer iki kez çalışır.

Observable Contract'ı: next, Sonra En Fazla Bir error ya da complete

RxJS dokümanı Observable'ın kurallarını bir regular expression ile tanımlar: next*(error|complete)?. Yani sıfır ya da daha fazla next, ardından en fazla bir kez ya error ya da complete. İkisinden biri geldiyse o execution bitmiştir; sonrasında gönderilen hiçbir şey consumer'a ulaşmaz. Bir sipariş durumu akışıyla deneyelim:

import { Observable } from "rxjs";

const orderStatus$ = new Observable<string>((subscriber) => {
  subscriber.next("Pending");
  subscriber.next("Paid");
  subscriber.error(new Error("payment provider timeout"));
  subscriber.next("Shipped");
  subscriber.complete();
  console.log("producer: closed =", subscriber.closed);
  return () => console.log("teardown");
});

orderStatus$.subscribe({
  next: (status) => console.log("next:", status),
  error: (err: Error) => console.log("error:", err.message),
  complete: () => console.log("complete"),
});

next: Pending
next: Paid
error: payment provider timeout
producer: closed = true
teardown

Hatadan sonraki "Shipped" ve complete() sessizce yutuldu; exception da fırlamadı. RxJS bunu producer'a verdiği Subscriber nesnesiyle sağlıyor: error ya da complete çağrıldığı anda closed true oluyor ve sonraki çağrılar hiçbir şey yapmıyor. Producer'ın subscriber.closed'a bakıp gereksiz işi bırakması da mümkün; uzun bir döngüde bu kontrol işe yarar.

Son satır da önemli: teardown, producer fonksiyonu döner dönmez çalıştı. Execution producer'ın içinde zaten bittiği için RxJS, dönen teardown fonksiyonunu alır almaz çağırıyor. Yani teardown'ın ne zaman çalışacağını "unsubscribe edilince" diye ezberlemek eksik kalıyor; doğrusu "execution bittiği an": error, complete ya da unsubscribe, hangisi önce gelirse.

Lazy Çalışma: Her subscribe Producer'ı Yeniden Çalıştırır

Lazy çalışmanın en pahalı sonucu HTTP'de ortaya çıkar. Snapshot'taki api/server.ts, Bookstore API'sinin küçük bir taklidi: her isteği numarasıyla loglayıp 200 ms sonra cevap veriyor. Stok bilgisini çeken bir Observable yazalım; fetch'i producer'ın içinde başlatıyoruz, teardown'da da AbortController ile iptal ediyoruz:

import { Observable } from "rxjs";

const apiUrl = "http://localhost:5180/api";

interface StockInfo {
  isbn: string;
  stock: number;
}

function getStock(isbn: string): Observable<StockInfo> {
  return new Observable<StockInfo>((subscriber) => {
    const controller = new AbortController();
    console.log(`GET /books/${isbn}/stock`);

    fetch(`${apiUrl}/books/${isbn}/stock`, { signal: controller.signal })
      .then((response) => response.json() as Promise<StockInfo>)
      .then((info) => {
        subscriber.next(info);
        subscriber.complete();
      })
      .catch((err) => subscriber.error(err));

    // teardown: unsubscribe ya da complete sonrası çalışır
    return () => controller.abort();
  });
}

const duneStock$ = getStock("9780441172719");

// iki farklı yer aynı Observable'a subscribe oluyor
duneStock$.subscribe((info) => console.log("badge:", info.stock));
duneStock$.subscribe((info) => console.log("cart warning:", info.stock < 3));

// üçüncü subscriber cevabı beklemeden vazgeçiyor
const subscription = duneStock$.subscribe((info) => console.log("never printed", info));
setTimeout(() => subscription.unsubscribe(), 50);

Bir terminalde node api/server.ts, diğerinde node src/01/04-stock-request.ts çalıştırınca istemci tarafı şunu yazıyor:

GET /books/9780441172719/stock
GET /books/9780441172719/stock
GET /books/9780441172719/stock
badge: 12
cart warning: false

Sunucu logu da aynı şeyi söylüyor:

[api] listening on http://localhost:5180
[api] #1 GET /api/books/9780441172719/stock
[api] #2 GET /api/books/9780441172719/stock
[api] #3 GET /api/books/9780441172719/stock
[api] #3 client aborted
[api] #1 200
[api] #2 200

Kodda tek bir duneStock$ var ama sunucuya üç istek gitti. Üçüncüsü 50 ms sonra unsubscribe edildi, teardown abort() çağırdı ve sunucu bağlantının kapandığını gördü. Bu davranış bir hata değil, Observable'ın tanımı: duneStock$ bir istek değil, "istek atma tarifi". Angular'ın HttpClient'ı da aynı şekilde çalışıyor; Angular dokümanı HttpClient Observable'larının cold olduğunu, aynı Observable'a birden çok subscribe olmanın birden çok backend isteği ürettiğini açıkça yazıyor. Template'te aynı Observable'a iki async pipe bağlamak bu yüzden iki istek demek. Tek isteği birden çok subscriber arasında paylaşmak bu serinin multicasting yazısının (share, shareReplay) konusu; cold ve hot ayrımına bir sonraki yazıda bakacağız.

Teardown: Execution Bitince Ne Temizlenir?

Gerçek producer'lar timer, event listener, WebSocket ya da HTTP bağlantısı gibi kaynaklar açar. Teardown fonksiyonu, o kaynağı kapatmanın tek doğru yeri. Stok değişikliklerini taklit eden bir producer yazalım: her 100 ms'de bir sipariş gelmiş gibi stok bir azalıyor, sıfıra inince akış tamamlanıyor.

import { Observable } from "rxjs";

// Sahte bir producer: Dune'un stoğu her 100 ms'de bir azalıyor (sipariş geliyormuş gibi)
function stockChanges(title: string, start: number): Observable<number> {
  return new Observable<number>((subscriber) => {
    let stock = start;
    console.log(`[${title}] timer started`);
    const id = setInterval(() => {
      stock -= 1;
      subscriber.next(stock);
      if (stock === 0) subscriber.complete();
    }, 100);

    return () => {
      clearInterval(id);
      console.log(`[${title}] timer cleared`);
    };
  });
}

// 1) consumer erken vazgeçiyor: teardown unsubscribe ile çalışır
const sub = stockChanges("Dune", 12).subscribe((s) => console.log("Dune stock:", s));
setTimeout(() => sub.unsubscribe(), 350);

// 2) producer kendisi bitiriyor: unsubscribe çağrılmadan teardown çalışır
setTimeout(() => {
  stockChanges("Middlemarch", 3).subscribe({
    next: (s) => console.log("Middlemarch stock:", s),
    complete: () => console.log("Middlemarch: sold out"),
  });
}, 500);

[Dune] timer started
Dune stock: 11
Dune stock: 10
Dune stock: 9
[Dune] timer cleared
[Middlemarch] timer started
Middlemarch stock: 2
Middlemarch stock: 1
Middlemarch stock: 0
Middlemarch: sold out
[Middlemarch] timer cleared

İki senaryoda da interval temizlendi. İkincisinde kimse unsubscribe çağırmadı; complete() execution'ı bitirdiği için RxJS teardown'ı kendisi çalıştırdı. Bu yüzden tamamlanan bir Observable'ı (tek cevaplı bir HTTP isteği gibi) ayrıca unsubscribe etmek zorunlu değil, tamamlanmayan bir Observable'ı (interval, event, WebSocket) unsubscribe etmek ise zorunlu.

Peki teardown hiç döndürülmezse? Aynı producer'ı teardown'sız yazalım:

import { Observable } from "rxjs";

let ticks = 0;

const leakyStock$ = new Observable<number>((subscriber) => {
  setInterval(() => {
    ticks++;
    subscriber.next(ticks);
  }, 100);
  // teardown dönmüyor: bu interval'ı kimse durduramaz
});

const sub = leakyStock$.subscribe((n) => console.log("value", n));
setTimeout(() => {
  sub.unsubscribe();
  console.log("unsubscribed at tick", ticks);
}, 250);
setTimeout(() => {
  console.log("ticks one second later:", ticks);
  process.exit(0);
}, 1250);

value 1
value 2
unsubscribed at tick 2
ticks one second later: 12

unsubscribe'tan sonra ekrana değer gelmedi; bu, kodun doğru çalıştığı izlenimini veriyor. Ama sayaç 2'den 12'ye çıktı: interval arka planda çalışmaya devam ediyor, subscriber.next çağrıları kapalı subscriber'da boşa gidiyor. Son satırdaki process.exit(0) da bu yüzden var; o satır olmasa Node süreci hiç kapanmaz. Bir Angular component'inde bu, sayfadan çıkıldıktan sonra da çalışan bir timer, her girişte bir tane daha.

40 Satırda Kendi Observable'ın

Buraya kadar gördüğümüz her şeyi (lazy çalışma, contract, teardown, pipe) kendimiz yazabiliriz. Aşağıdaki class RxJS'in kodu değil, onun fikrini gösteren küçük bir model; Observable kısmı 40 satırın altında, map ve filter ile birlikte dosya 46 satır:

export interface Observer<T> {
  next: (value: T) => void;
  error: (err: unknown) => void;
  complete: () => void;
}
type Teardown = (() => void) | void;
type Operator<A, B> = (source: MiniObservable<A>) => MiniObservable<B>;

export class MiniObservable<T> {
  readonly producer: (subscriber: Observer<T>) => Teardown;
  constructor(producer: (subscriber: Observer<T>) => Teardown) {
    this.producer = producer;
  }

  subscribe(partial: Partial<Observer<T>>): { unsubscribe: () => void } {
    let closed = false;
    let teardown: Teardown = undefined;
    const close = () => {
      if (closed) return;
      closed = true;
      teardown?.();
    };
    const subscriber: Observer<T> = {
      next: (v) => { if (!closed) partial.next?.(v); },
      error: (e) => { if (!closed) { partial.error?.(e); close(); } },
      complete: () => { if (!closed) { partial.complete?.(); close(); } },
    };
    try {
      teardown = this.producer(subscriber);
    } catch (err) {
      subscriber.error(err);
    }
    if (closed) teardown?.(); // producer senkron olarak bitirdi
    return { unsubscribe: close };
  }

  pipe(...operators: Operator<any, any>[]): MiniObservable<any> {
    return operators.reduce((source, op) => op(source), this as MiniObservable<any>);
  }
}

export const map = <A, B>(fn: (v: A) => B): Operator<A, B> => (source) =>
  new MiniObservable<B>((sub) => source.subscribe({ ...sub, next: (v) => sub.next(fn(v)) }).unsubscribe);

export const filter = <A>(fn: (v: A) => boolean): Operator<A, A> => (source) =>
  new MiniObservable<A>((sub) => source.subscribe({ ...sub, next: (v) => fn(v) && sub.next(v) }).unsubscribe);

Parçaları tek tek eşleştirelim. Constructor yalnız producer fonksiyonunu saklıyor, çalıştırmıyor: lazy çalışma bu kadar. subscribe her çağrıda yeni bir closed bayrağı ve yeni bir subscriber nesnesi kuruyor, sonra producer'ı çalıştırıyor: her subscribe'ın kendi execution'ı olması bu. closed kontrolleri contract'ın kendisi: error ya da complete'ten sonra hiçbir çağrı consumer'a ulaşmıyor. close teardown'ı bir kez çalıştırıyor; producer senkron bittiyse teardown, dönüş değeri alınır alınmaz çağrılıyor (RxJS'te gördüğümüz "teardown" satırının sebebi). Producer'ın fırlattığı exception try/catch ile error bildirimine çevriliyor.

Operator'ler de sandığından basit. map(fn), bir Observable alıp yenisini döndüren bir fonksiyon üretiyor. Yeni Observable'a subscribe olunca o da kaynağa subscribe oluyor, her değeri fn'den geçirip aşağı iletiyor ve kaynağın unsubscribe'ını kendi teardown'ı olarak döndürüyor. Böylece zincirin en altındaki unsubscribe, en üstteki producer'a kadar ulaşıyor. pipe ise bu fonksiyonları sırayla uygulayan bir reduce. Deneyelim:

import { MiniObservable, map, filter } from "./mini-observable.ts";

const stock$ = new MiniObservable<number>((subscriber) => {
  console.log("producer started");
  let stock = 5;
  const id = setInterval(() => {
    stock -= 1;
    subscriber.next(stock);
    if (stock === 0) subscriber.complete();
  }, 100);
  return () => {
    clearInterval(id);
    console.log("teardown");
  };
});

const lowStock$ = stock$.pipe(
  filter((s: number) => s < 3),
  map((s: number) => `only ${s} left`),
);

console.log("nothing has run yet");
lowStock$.subscribe({
  next: (msg) => console.log(msg),
  complete: () => console.log("complete"),
});

nothing has run yet
producer started
only 2 left
only 1 left
only 0 left
complete
teardown

Yazının başındaki tahmin kodunu MiniObservable ile çalıştırınca (snapshot'ta 08-mini-predict.ts) RxJS ile aynı sırayı alıyoruz: "producer started" iki kez, 10 yok, "after subscribe" en sonda. Producer'da fırlatılan bir exception da error callback'ine düşüyor:

before subscribe
producer started
A got 12
A got 11
teardown
producer started
B got 12
B got 11
teardown
after subscribe
error: stock service unavailable

RxJS'in bu modele eklediği şeyler de belli: birden çok teardown'ı toplayan Subscription nesnesi, yüzden fazla test edilmiş operator, zamanı yöneten scheduler'lar, tip güvenli pipe overload'ları (bizimki any'ye düşüyor) ve yakalanmamış hataların raporlanması. Sonuncusu bir sonraki bölümün konusu.

Üretimde Dikkat

Bookstore'un sepet sayfasında API, bir kitabın fiyatını null döndürdü diyelim. Producer'ı try/catch ile sarmışız, error callback'i de yazmışız:

import { Observable } from "rxjs";

const cart$ = new Observable<{ title: string; price: number | null }>((subscriber) => {
  try {
    subscriber.next({ title: "Dune", price: 450 });
    subscriber.next({ title: "Emma", price: null }); // API cevabında fiyat yok
    subscriber.next({ title: "The Hobbit", price: 390 });
    subscriber.complete();
  } catch (err) {
    console.log("producer caught:", err); // hiç çalışmaz
    subscriber.error(err);
  }
});

console.log("before subscribe");
cart$.subscribe({
  next: (item) => console.log(item.title, item.price!.toFixed(2)),
  error: (err) => console.log("error callback:", err), // hiç çalışmaz
  complete: () => console.log("complete"),
});
console.log("after subscribe");

before subscribe
Dune 450.00
The Hobbit 390.00
complete
after subscribe
C:\Users\user\bookstore-rxjs\node_modules\rxjs\dist\cjs\internal\util\reportUnhandledError.js:13
            throw err;
            ^

TypeError: Cannot read properties of null (reading 'toFixed')
    at Object.next (file:///C:/Users/user/bookstore-rxjs/src/01/09-lost-error.ts:17:55)

Emma satırı sessizce kayboldu, akış The Hobbit ile devam edip complete oldu; ne producer'ın catch'i ne de error callback'i çalıştı. RxJS 7, consumer'ın next callback'inde fırlayan exception'ı producer'a geri göndermiyor; onu yakalayıp bir sonraki macrotask'ta yeniden fırlatıyor. Node'da bu yakalanmamış exception süreci kapattı. Aynı kodu Chromium'da çalıştırdığımda sayfa açık kaldı, konsolda yalnız bir uncaught error vardı:

[console.log] before subscribe
[console.log] Dune 450.00
[console.log] The Hobbit 390.00
[console.log] complete
[console.log] after subscribe
[pageerror] Cannot read properties of null (reading 'toFixed')
[console.log] page still alive

Kullanıcı açısından sonuç: sepette bir kitap görünmüyor, toplam yanlış ve hata yalnız konsolda. Buradan iki pratik kural çıkıyor. Birincisi, subscribe callback'ine iş mantığı koyma; dönüşümü map gibi operator'lerde yap. Operator içinde fırlayan exception, contract'a uygun bir error bildirimine dönüşür ve hata yönetimi operator'leriyle yakalanabilir (bu serinin error handling yazısında). İkincisi, kimsenin yakalamadığı hataları merkezi olarak loglamak için RxJS'in config.onUnhandledError ayarını kullan:

import { config, Observable } from "rxjs";

// hiçbir subscriber'ın yakalamadığı hatalar için tek merkez
config.onUnhandledError = (err) => {
  console.log("[log] unhandled RxJS error:", (err as Error).message);
};

const price$ = new Observable<number>((subscriber) => {
  subscriber.error(new Error("price service returned 503"));
});

price$.subscribe((price) => console.log("price", price)); // error callback yok
price$.subscribe({
  next: (price) => console.log("price", price),
  error: (err: Error) => console.log("handled here:", err.message),
});
console.log("after subscribe");

handled here: price service returned 503
after subscribe
[log] unhandled RxJS error: price service returned 503

error callback'i olmayan ilk subscription'ın hatası kaybolmadı, ama yine asenkron raporlandı ("after subscribe"tan sonra). Production'da bu hook'u Bookstore'un log altyapısına bağlamak, "konsolda kalan" hataları görünür kılar. Ölçüm tarafında ise iki sayı takip etmeye değer: tarayıcının Network sekmesinde bir sayfa açılışında aynı URL'e kaç istek gittiği (lazy çalışmanın yan etkisi) ve uzun süre açık kalan sayfada bellek ile timer sayısının artıp artmadığı (teardown eksikliği).

Sık Düşülen Tuzaklar

• Observable'ı "bir kez çalışan istek" sanmak. Her subscribe producer'ı baştan çalıştırır; aynı HTTP Observable'ına iki yerden subscribe olmak iki istek demektir.
• Observable'ın her zaman asenkron olduğunu varsaymak. Producer senkron next çağırıyorsa değerler subscribe dönmeden teslim edilir; "subscribe'dan sonraki satır önce çalışır" varsayımına dayanan kod sessizce bozulur.
• new Observable içinde kaynak açıp teardown döndürmemek. unsubscribe sonrası ekrana değer gelmemesi kaynağın kapandığını göstermez; interval, listener ya da bağlantı açık kalır.
• complete ya da error'dan sonra next çağırıp değerin gideceğini sanmak. Contract gereği hepsi yutulur, hata da alınmaz; producer'da subscriber.closed'a bakmak gereksiz işi önler.
• İş mantığını subscribe callback'ine yazmak. Orada fırlayan exception error callback'ine düşmez, asenkron olarak yeniden fırlatılır ve akışın geri kalanı devam eder.

Küçük Bir Deneme

★☆☆ 04-stock-request.ts'teki getStock'u Emma için (9780141439587) üç kez çağır ve dönen üç Observable'dan yalnız birine subscribe ol. Sunucu logunda kaç istek görüyorsun? Sonra o tek Observable'a setTimeout ile 1 saniye arayla bir kez daha subscribe ol; kaç istek gideceğini önce tahmin et, sonra logla doğrula.

★★☆ MiniObservable'a bir take(n) operator'ü ekle: ilk n değeri iletsin, sonra complete göndersin ve kaynağa unsubscribe olsun. stockChanges benzeri bir interval producer'ıyla dene; "teardown" satırının complete'ten hemen sonra çıktığını doğrula. İpucu: kaynağın subscription'ı henüz dönmemişken senkron bir kaynak n'e ulaşabilir.

★★★ getStock'u, sunucu 503 dönerse error bildirimi gönderecek şekilde değiştir (response.ok kontrolü). Ardından config.onUnhandledError'ı, hatanın hangi ISBN (kitabın uluslararası standart numarası) için olduğunu da loglayacak şekilde ayarla. Hata nesnesine bilgiyi nasıl taşıdığını ve bu bilginin neden subscribe callback'inde değil producer'da eklenmesi gerektiğini not et.

Bu yazıda Observable'ı hep new Observable ile elle kurduk. Gerçek kodda bunu nadiren yaparsın; RxJS'in of, from, defer, timer, fromEvent gibi creation function'ları aynı işi hazır yapıyor. Bir sonraki yazıda bu function'ları ve "cold" ile "hot" Observable ayrımını göreceğiz; from(promise)'in neden lazy olmadığını ve retry'da isteği yeniden atmak için neden defer gerektiğini de sunucu loguyla ölçeceğiz.