RxJS'te Custom Operator Yazmak ve Okunur Kalmak: pipe Kompozisyonu, Lint Kuralları ve Debug

Sürüm notu: Bu yazıdaki örnekler RxJS 7.8.2, TypeScript 6.0, Node.js 24, Angular 22.2, ESLint 10.15, typescript-eslint 8.72 ve eslint-plugin-rxjs-x 1.0.6 ile denenmiştir (Angular projesi, Angular 22.2'nin desteklediği TypeScript 6.0'ı kullanıyor); Angular örnekleri Playwright 1.65 ile headless Chromium'da, snapshot'taki sahte Bookstore API'sine karşı çalıştırılmıştır.

Önce Tahmin Et

Sitedeki serilerin ortak örnek uygulaması Bookstore'da (bir online kitapçı) bir kitabın stoğu saniyede bir sorgulanıyor ve ekranda yalnız stok değiştiğinde güncelleniyor. Ekipten biri "aynı değeri iki kez geçirme" işini kendi operator'ü olarak yazmış ve sayfa yalnız ilk iki değişikliği gösterip duruyor:

import { map, Observable, take, tap, timer, type MonoTypeOperatorFunction } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// elle yazılmış bir operator: stok değerini yalnız öncekinden farklıysa geçirir
function stockChanges(): MonoTypeOperatorFunction<number> {
  return (source) =>
    new Observable<number>((subscriber) => {
      let previous: number | undefined;
      source.subscribe({
        next: (stock) => {
          if (stock !== previous) subscriber.next(stock);
          previous = stock;
        },
        error: (err: unknown) => subscriber.error(err),
        complete: () => subscriber.complete(),
      });
    });
}

// API'nin her sorguda bildirdiği Dune stoğu
const answers = [12, 12, 11, 11, 11, 10, 10, 9];

runInVirtualTime("first two stock changes of Dune", () => {
  timer(0, 1000)
    .pipe(
      tap((i) => log(`-> GET /books/9780441172719/stock (poll ${i + 1})`)),
      map((i) => answers[i % answers.length]),
      stockChanges(),
      take(2),
    )
    .subscribe({
      next: (stock) => log(`screen: ${stock} in stock`),
      complete: () => log("done"),
    });
}, 5000);

Ekranda "done" yazdıktan sonra kaç istek daha gider? take(2) tamamlandığında zincirin en başındaki timer'a ne olur?

Seri boyunca bazı pipe'ları birkaç kez yazdık: debounce'lu arama bloğu, backoff'lu retry, bir isteğin etrafındaki loading durumu. Bu yazıda bunları isimli, tipli ve test edilebilir operator'lere çeviriyoruz: önce pipe() ile mevcut operator'lerden, sonra gerektiğinde new Observable ile sıfırdan. Ardından uzun pipe'ları okunur tutmanın kurallarına ve bu kuralların bir kısmını ESLint'e bırakmaya bakıyoruz.

Operator Sadece Bir Fonksiyon

Marble diagram yazısında gördüğümüz tanımı hatırla: pipeable operator, bir Observable alıp yeni bir Observable döndüren fonksiyon. RxJS bunun tipini OperatorFunction<T, R> olarak veriyor; girdi ve çıktı tipi aynıysa kısaltması MonoTypeOperatorFunction<T>. Kendi operator'ünü yazmak, bu tipte bir değer döndüren bir fonksiyon yazmak demek. İki yolu var: mevcut operator'leri pipe() fonksiyonuyla birleştirmek (RxJS dokümanı da önce bunu öneriyor ve işlerin çoğu böyle çözülüyor) ya da mevcut operator'lerle ifade edilemeyen bir davranış için (source) => new Observable(subscriber => …) ile sıfırdan yazmak.

Tahmin örneğindeki stockChanges ikinci yoldan yazılmış ve bir şeyi unutmuş. Çalıştırınca:

--- first two stock changes of Dune
    0 ms  -> GET /books/9780441172719/stock (poll 1)
    0 ms  screen: 12 in stock
 1000 ms  -> GET /books/9780441172719/stock (poll 2)
 2000 ms  -> GET /books/9780441172719/stock (poll 3)
 2000 ms  screen: 11 in stock
 2000 ms  done
 3000 ms  -> GET /books/9780441172719/stock (poll 4)
 4000 ms  -> GET /books/9780441172719/stock (poll 5)
 5000 ms  -> GET /books/9780441172719/stock (poll 6)

take(2) ikinci değişiklikte tamamlandı ve ekran "done" yazdı, ama istekler sürüyor. Sanal saati 5. saniyede biz durdurduk; gerçek uygulamada polling sayfa kapanana kadar, root bir service'teyse uygulama kapanana kadar devam ederdi. Sebep, new Observable'a verilen fonksiyonun hiçbir şey döndürmemesi. take(2) tamamlanınca kendi kaynağından, yani stockChanges'in döndürdüğü Observable'dan unsubscribe oluyor. Bu Observable'ın teardown'u (unsubscribe ya da tamamlanma anında çalışan temizlik fonksiyonu) yok; içerideki source.subscribe(…)'un Subscription'ı hiçbir yere bağlanmadığı için timer'a unsubscribe sinyali hiç ulaşmıyor.

Düzeltme tek kelime: içerideki Subscription'ı döndürmek. new Observable'a verilen fonksiyon bir teardown fonksiyonu ya da doğrudan bir Subscription döndürebilir:

// elle yazılmış bir operator: stok değerini yalnız öncekinden farklıysa geçirir
function stockChanges(): MonoTypeOperatorFunction<number> {
  return (source) =>
    new Observable<number>((subscriber) => {
      let previous: number | undefined;
      // Subscription'ı döndürmek onu teardown yapar: unsubscribe artık kaynağa ulaşır
      return source.subscribe({
        next: (stock) => {
          if (stock !== previous) subscriber.next(stock);
          previous = stock;
        },
        error: (err: unknown) => subscriber.error(err),
        complete: () => subscriber.complete(),
      });
    });
}

--- first two stock changes of Dune, with teardown
    0 ms  -> GET /books/9780441172719/stock (poll 1)
    0 ms  screen: 12 in stock
 1000 ms  -> GET /books/9780441172719/stock (poll 2)
 2000 ms  -> GET /books/9780441172719/stock (poll 3)
 2000 ms  screen: 11 in stock
 2000 ms  done

Üçüncü istekten sonra hiçbir şey yok. RxJS dokümanı sıfırdan yazılan bir operator için üç şartı sayıyor: kaynağa subscribe olurken next, error ve complete'in üçünü de ele almak, temizliği yapan bir fonksiyon yazmak ve bu fonksiyonu new Observable'a verilen fonksiyondan döndürmek. Bu örnekte üçüncüsü eksikti. Bu arada stockChanges'in yaptığı iş zaten distinctUntilChanged()'in kendisi; sıfırdan yazmadan önce mevcut bir operator'ün aynı işi yapıp yapmadığına bakmak en ucuz kontrol.

pipe() ile Mevcut Operator'lerden Yeni Operator

Arama kutusunda üç adımı her seferinde yan yana yazıyorduk: debounceTime, trim ve distinctUntilChanged. rxjs'ten gelen bağımsız pipe() fonksiyonu, Observable'ın .pipe() method'unun aksine bir kaynak almıyor; verdiğin operator'leri birleştirip tek bir OperatorFunction döndürüyor:

// 1. kullanıcının yazmayı bıraktığı arama terimi: bekle, kırp, tekrarları atla
export function settledTerm(dueTimeMs = 300): MonoTypeOperatorFunction<string> {
  return pipe(
    debounceTime(dueTimeMs),
    map((term) => term.trim()),
    distinctUntilChanged(),
  );
}

Bunu bir sonraki adımla birleştirince seri boyunca yazdığımız arama bloğu tek isimli bir operator oluyor. İçerideki isteğe birazdan sıfırdan yazacağımız debugTap'i taktık, böylece switchMap'in her isteğe ne zaman subscribe olup ne zaman bıraktığı görünüyor:

import { concat, map, NEVER, pipe, switchMap, timer, type Observable, type OperatorFunction } from "rxjs";
import { books } from "../../data/books.ts";
import { emitAt, log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { debugTap, settledTerm } from "./operators.ts";

// sahte bir GET /books?q=…: 500 ms sonra cevap verir
function findBooks(term: string): Observable<string[]> {
  return timer(500).pipe(
    map(() => books.filter((b) => b.title.toLowerCase().includes(term)).map((b) => b.title)),
  );
}

// önceki yazılardaki arama bloğunun tamamı, artık tek isimli bir operator
function searchBooks(): OperatorFunction<string, string[]> {
  return pipe(
    settledTerm(300),
    switchMap((term) => findBooks(term).pipe(debugTap(`GET /books?q=${term}`, log))),
  );
}

runInVirtualTime("search box", () => {
  // kullanıcının yazdıkları; NEVER, gerçek bir metin kutusu gibi input'u açık tutar
  const input$ = concat(
    emitAt([0, "d"], [100, "du"], [200, "dun"], [300, "dune"], [1200, "dune "], [1600, "em"], [2000, "emma"]),
    NEVER,
  );
  input$.pipe(searchBooks()).subscribe((titles) => log(`screen: ${titles.join(", ")}`));
});

--- search box
  600 ms  [GET /books?q=dune] subscribe
 1100 ms  [GET /books?q=dune] next ["Dune"]
 1100 ms  screen: Dune
 1100 ms  [GET /books?q=dune] complete
 1900 ms  [GET /books?q=em] subscribe
 2300 ms  [GET /books?q=em] unsubscribe
 2300 ms  [GET /books?q=emma] subscribe
 2800 ms  [GET /books?q=emma] next ["Emma"]
 2800 ms  screen: Emma
 2800 ms  [GET /books?q=emma] complete

"dune" son tuştan 300 ms sonra arandı. 1200. ms'de eklenen boşluk trim'den sonra aynı terimi verdi ve distinctUntilChanged onu durdurdu. "em" isteği sürerken "emma" oturdu ve switchMap eski isteği bırakıp yenisini başlattı; log'daki unsubscribe satırı tam bu an. Kullanıcının yazmayı bıraktığı terimin artık bir adı var ve bookstore-web'deki iki arama component'i de onu kullanıyor: search-rxjs.ts ve search-interop.ts içindeki üç satırlık blok settledTerm(300) oldu. Arama sayfasını önceki yazıdaki Playwright testiyle yeniden çalıştırdık; üç kutu da "dune" için yine tek istek gönderdi.

retryWithBackoff: Seçenek Alan Bir Operator

Error handling yazısındaki backoff'lu retry iki yerde yaşıyordu: örnek projede retryDelay yardımcı fonksiyonu, bookstore-web'de interceptor içinde bir kopyası. İkisi de retry({ count, delay })'e verilen bir fonksiyondu; operator olarak adı ve seçenekleri belli oluyor:

// 2. exponential backoff ve jitter ile retry, yalnız geçebilecek hatalar için
export interface BackoffOptions {
  count?: number;
  baseMs?: number;
  maxMs?: number;
  isRetryable?: (err: unknown) => boolean;
  random?: () => number;
  onRetry?: (err: unknown, retryCount: number, delayMs: number) => void;
}

export function retryWithBackoff<T>(options: BackoffOptions = {}): MonoTypeOperatorFunction<T> {
  const { count = 3, baseMs = 500, maxMs = 8000, isRetryable = () => true, random = Math.random, onRetry } = options;
  return retry({
    count,
    delay: (err: unknown, retryCount) => {
      if (!isRetryable(err)) return throwError(() => err);
      const delayMs = backoffDelay(retryCount, random, baseMs, maxMs);
      onRetry?.(err, retryCount, delayMs);
      return timer(delayMs);
    },
  });
}

Varsayılanlar kasıtlı olarak tutucu: üç deneme, 500 ms taban, en fazla 8 saniye. isRetryable varsayılan olarak her hatayı tekrar deniyor, çünkü operator hangi hatanın geçici olduğunu bilemez; HTTP için bunu çağıran taraf veriyor. random seçeneği testte sabit bir sayı üreteci vermek için, onRetry ise log için. Error handling yazısındaki senaryoyu aynı seed ile çalıştırdık:

import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { isRetryable, seededRandom } from "../09/backoff.ts";
import { flakyHttp } from "../09/fake-http.ts";
import { retryWithBackoff } from "./operators.ts";

runInVirtualTime("503, 503, then 200", () => {
  flakyHttp("GET /books/9780441172719/stock", 200, [503, 503, 200], () => 12)
    .pipe(
      retryWithBackoff({
        count: 4,
        isRetryable,
        random: seededRandom(42),
        onRetry: (err, retryCount, delayMs) => log(`retry ${retryCount} in ${delayMs} ms (${String(err)})`),
      }),
    )
    .subscribe({
      next: (stock) => log(`stock: ${stock}`),
      error: (err: Error) => log(`stock failed: ${err.message}`),
    });
});

runInVirtualTime("404 is not retried", () => {
  flakyHttp("GET /books/9780000000000", 200, [404], () => "book")
    .pipe(retryWithBackoff({ count: 4, isRetryable }))
    .subscribe({
      next: (book) => log(`book: ${book}`),
      error: (err: Error) => log(`book failed: ${err.message}`),
    });
});

--- 503, 503, then 200
    0 ms  -> GET /books/9780441172719/stock
  200 ms  <- GET /books/9780441172719/stock 503
  200 ms  retry 1 in 400 ms (HttpError: HTTP 503)
  600 ms  -> GET /books/9780441172719/stock
  800 ms  <- GET /books/9780441172719/stock 503
  800 ms  retry 2 in 724 ms (HttpError: HTTP 503)
 1524 ms  -> GET /books/9780441172719/stock
 1724 ms  <- GET /books/9780441172719/stock 200
 1724 ms  stock: 12
--- 404 is not retried
    0 ms  -> GET /books/9780000000000
  200 ms  <- GET /books/9780000000000 404
  200 ms  book failed: HTTP 404

Bekleme süreleri (400 ve 724 ms) error handling yazısındakilerle birebir aynı; operator davranışı değiştirmeden yalnızca bir isim kazandırdı. 404 tekrar denenmedi. Interceptor artık aynı operator'ü kullanıyor ve yalnız HTTP'ye özgü kararları veriyor:

export const errorInterceptor: HttpInterceptorFn = (req, next) => {
  const serverStatus = inject(ServerStatus);
  // yalnız GET'i tekrarlamak güvenli; bir POST ikinci bir sipariş oluşturabilir
  const retries = req.method === 'GET' ? 3 : 0;

  return next(req).pipe(
    retryWithBackoff({
      count: retries,
      isRetryable,
      onRetry: (err, retryCount) => {
        const status = err instanceof HttpErrorResponse ? err.status : err;
        console.warn(`[http] ${req.method} ${req.url} failed (${status}), retry ${retryCount}`);
      },
    }),

loadingState: Template İçin Durum Üretmek

Bir isteğin etrafında üç durum var: bekleniyor, geldi, hata verdi. Bunu her component'te ayrı signal'larla tutmak yerine akışın kendisini bu durumlara çeviren bir operator yazıyoruz. Dönüş tipi bir discriminated union (ortak bir alanla, burada status, ayırt edilen tip birleşimi); TypeScript status'a bakınca data ya da error'a erişimi daraltıyor:

// 3. bir isteği template'in gösterebileceği durumlara çevir: loading, sonra success ya da error
export type LoadingState<T> =
  | { status: "loading" }
  | { status: "success"; data: T }
  | { status: "error"; error: unknown };

export function loadingState<T>(): OperatorFunction<T, LoadingState<T>> {
  return (source) =>
    source.pipe(
      map((data): LoadingState<T> => ({ status: "success", data })),
      startWith<LoadingState<T>>({ status: "loading" }),
      catchError((error: unknown) => of<LoadingState<T>>({ status: "error", error })),
    );
}

Sıra önemli. startWith, map'ten sonra geldiği için ilk değer olarak loading veriliyor; catchError en sonda olduğu için hata da bir duruma dönüşüyor ve akış error ile değil complete ile bitiyor. Bu operator'ü retryWithBackoff ile birlikte kullanırken de sıra önemli:

import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { isRetryable, seededRandom } from "../09/backoff.ts";
import { flakyHttp } from "../09/fake-http.ts";
import { loadingState, retryWithBackoff, type LoadingState } from "./operators.ts";

const show = (state: LoadingState<string[]>) =>
  log(
    state.status === "success" ? `success: ${state.data.join(", ")}`
    : state.status === "error" ? `error: ${String(state.error)}`
    : "loading",
  );

runInVirtualTime("categories: one 503, then 200", () => {
  flakyHttp("GET /categories", 200, [503, 200], () => ["classics", "fantasy"])
    .pipe(retryWithBackoff({ count: 2, isRetryable, random: seededRandom(42) }), loadingState())
    .subscribe(show);
});

runInVirtualTime("categories: 503 every time", () => {
  flakyHttp("GET /categories", 200, [503], () => ["classics"])
    .pipe(retryWithBackoff({ count: 2, isRetryable, random: seededRandom(42) }), loadingState())
    .subscribe(show);
});

runInVirtualTime("wrong order: loadingState before retryWithBackoff", () => {
  flakyHttp("GET /categories", 200, [503, 200], () => ["classics", "fantasy"])
    .pipe(loadingState(), retryWithBackoff({ count: 2, isRetryable, random: seededRandom(42) }))
    .subscribe(show);
});

--- categories: one 503, then 200
    0 ms  loading
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  600 ms  -> GET /categories
  800 ms  <- GET /categories 200
  800 ms  success: classics, fantasy
--- categories: 503 every time
    0 ms  loading
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  600 ms  -> GET /categories
  800 ms  <- GET /categories 503
 1524 ms  -> GET /categories
 1724 ms  <- GET /categories 503
 1724 ms  error: HttpError: HTTP 503
--- wrong order: loadingState before retryWithBackoff
    0 ms  loading
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  200 ms  error: HttpError: HTTP 503

Doğru sırada ekran tek bir loading gösteriyor, retry'lar arka planda oluyor ve sonuç ya veri ya tek bir hata. Ters sırada loadingState hatayı ilk denemede bir değere çevirdi; retryWithBackoff'un önüne hiç hata gelmediği için hiçbir şey tekrar denenmedi. Kural basit: bir hatayı değere çeviren operator, o hatayı görmesi gereken her şeyden sonra gelmeli.

Katalog sayfası artık boş bir listeye düşmek yerine bu durumları gösteriyor. startWith ilk değeri senkron verdiği için toSignal'ı requireSync: true ile kullanabiliyoruz; tipte undefined yok. Angular template'i @switch içinde union'ı daraltıyor; @case ('success') içinde state.data'ya erişmek derleniyor, başka bir case'de derlenmezdi:

    @let state = books();
    @switch (state.status) {
      @case ('loading') {
        <p class="books-state">Loading books...</p>
      }
      @case ('error') {
        <p class="books-state">{{ errorMessage(state.error) }}</p>
      }
      @case ('success') {
        <ul>
          @for (book of state.data; track book.isbn) {
            <li>
              <a [routerLink]="['/books', book.isbn]">{{ book.title }}</a> - {{ book.price }} TRY
              <button type="button" (click)="cart.add(book)" [disabled]="book.stock === 0">
                Add to cart
              </button>
            </li>
          }
        </ul>
      }
    }
  `,
})
export class CatalogPage {
  protected readonly cart = inject(CartStore);
  // interceptor zaten tekrar denedi; sayfa loading, liste ya da bir mesaj gösterir
  protected readonly books = toSignal(
    inject(HttpClient).get<Book[]>(`${apiUrl}/books`).pipe(loadingState()),
    { requireSync: true },
  );
  protected readonly errorMessage = errorMessage;
  protected readonly categories$ = inject(CategoryService).categories$;
}

API'yi önce FAIL=/api/books:2 ile (ilk iki istek 503), sonra bütün istekler 503 dönecek şekilde başlatıp sayfayı Playwright ile açtık:

 0.1 s screen: Loading books...
 0.3 s GET /api/books 503
 0.3 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.3 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 1
 0.9 s GET /api/books 503
 0.9 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.9 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 2
 2.0 s GET /api/books 200
 2.0 s screen: 6 books: Dune, Emma, The Hobbit…

 0.1 s screen: Loading books...
 0.3 s GET /api/books 503
 0.3 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.3 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 1
 0.8 s GET /api/books 503
 0.8 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 0.8 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 2
 1.8 s GET /api/books 503
 1.8 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 1.8 s [console.warning] [http] GET http://localhost:5180/api/books failed (503), retry 3
 3.8 s GET /api/books 503
 3.8 s [console.error] Failed to load resource: the server responded with a status of 503 (Service Unavailable)
 3.8 s [console.error] [http] GET http://localhost:5180/api/books failed: 503
 3.8 s screen: The Bookstore service is not responding. Please try again.

İlkinde kullanıcı iki saniye boyunca "Loading books..." görüyor, sonra liste geliyor; ikincisinde dört denemeden sonra, 3.8. saniyede anlaşılır bir mesaj. Önceki sürümde ikinci durumda global banner'ın altında boş bir liste kalıyordu; kullanıcı kitapların mı bittiğini yoksa servisin mi çöktüğünü ayırt edemiyordu.

Sıfırdan Yazmak: debugTap

Seri boyunca bir akışın içinde ne olduğunu görmek için tap ve finalize ile log yazdık. tap değerleri, hatayı ve tamamlanmayı görüyor, finalize sonu görüyor, ama ikisi de bir subscription'ın ne zaman başladığını ve sonun bir unsubscribe mı yoksa bir complete mi olduğunu tek başına söylemiyor. Bunu mevcut operator'lerle birleştirmek mümkün ama okunması zor oluyor; sıfırdan yazmak daha net:

// 4. sıfırdan: akışın her event'ini logla, unsubscribe anı dahil
export function debugTap<T>(label: string, log: (line: string) => void = console.log): MonoTypeOperatorFunction<T> {
  return (source) =>
    new Observable<T>((subscriber) => {
      log(`[${label}] subscribe`);
      let finished = false;
      const subscription = source.subscribe({
        next: (value) => {
          log(`[${label}] next ${JSON.stringify(value)}`);
          subscriber.next(value);
        },
        error: (err: unknown) => {
          finished = true;
          log(`[${label}] error ${err instanceof Error ? err.message : String(err)}`);
          subscriber.error(err);
        },
        complete: () => {
          finished = true;
          log(`[${label}] complete`);
          subscriber.complete();
        },
      });
      // teardown: complete, error ve unsubscribe'da çalışır; kaynağı da durdurmalı
      return () => {
        if (!finished) log(`[${label}] unsubscribe`);
        subscription.unsubscribe();
      };
    });
}

RxJS dokümanındaki üç şartın hepsi burada: üç observer fonksiyonu da var, error ve complete aşağıya iletiliyor ve teardown kaynaktan unsubscribe oluyor. finished bayrağı, teardown'un complete ya da error'dan sonra da çalıştığı gerçeğini ayırt etmek için; böylece log'daki unsubscribe satırı yalnız gerçekten dışarıdan bırakılan bir subscription için yazılıyor. Arama örneğinde "em" isteğinin satırı buydu. log parametresi sanal zamanlı örneklerde zamanı da yazan bir fonksiyon vermek için; varsayılanı console.log.

Generic'ler: any'ye Düşmeden

Custom operator'lerde sık görülen bir kısayol, tipi OperatorFunction<any, any> yazıp geçmek. Aynı loadingState'i bir kez öyle, bir kez generic olarak kullanıp aynı yazım hatasını yaptık:

import { catchError, map, of, startWith, type OperatorFunction } from "rxjs";
import { books } from "../../data/books.ts";
import { loadingState } from "./operators.ts";

// aynı operator, kestirmeden tiplenmiş hâli
function loadingStateAny(): OperatorFunction<any, any> {
  return (source) =>
    source.pipe(
      map((data) => ({ status: "success", data })),
      startWith({ status: "loading" }),
      catchError((error) => of({ status: "error", error })),
    );
}

of(books).pipe(loadingStateAny()).subscribe((state) => {
  // property adında yazım hatası: derlenir, undefined yazar
  if (state.status === "success") console.log("any:", state.data.lenght);
});

of(books).pipe(loadingState()).subscribe((state) => {
  // @ts-expect-error -- generic sürüm aynı yazım hatasını derleme anında yakalar
  if (state.status === "success") console.log("generic:", state.data.lenght);
});

any: undefined

any'li sürüm derleniyor ve çalışma anında undefined yazıyor. Generic sürümde @ts-expect-error satırını silince tsc şunu söylüyor:

src/11/06-generic-any.ts(22,70): error TS2551: Property 'lenght' does not exist on type 'readonly Book[]'. Did you mean 'length'?

Dikkat edilecek bir nokta: generic tipi elle yazmadık. loadingState() argüman almıyor, ama TypeScript T'yi pipe'ın beklediği tipten çıkarıyor ve readonly Book[] buluyor. İlk denemede loadingState<Book[]>() yazmıştık ve derleyici, örnek veri readonly olduğu için başka bir hata verdi; çoğu zaman en doğru tip, çıkarımın bulduğu tip. @ts-expect-error burada bir test gibi çalışıyor: altındaki satır hata vermezse npm run check başarısız oluyor.

Okunur Pipe Kuralları ve ESLint

Custom operator'ün asıl faydası teknik değil, okunurluk. Code review'da işimize yarayan kurallar:

• tap yalnız side effect için: log, analytics, bir signal'ı set etmek. Değeri değiştiren ya da başka bir istek başlatan bir tap, aslında map ya da switchMap olmalı.
• Beş altı adımı geçen bir pipe'ı, her biri bir iş yapan isimli operator'lere böl. settledTerm, searchBooks, retryWithBackoff, loadingState okunduğunda ne yaptığı belli; on satırlık bir pipe'ta aynı bilgiyi çıkarmak zaman alıyor.
• İsim sonucu anlatsın, uygulamayı değil: debounceAndTrimAndDistinct değil settledTerm.
• Seçenekleri bir object ile al ve tutucu varsayılanlar ver; RxJS 7'nin retry, share ve timeout'u da bu yolu izliyor.
• subscribe içinde subscribe yazma; iki akışı bir higher-order mapping operator'ü birleştirir.

Son iki kuralı ve benzerlerini insan gözüne bırakmak zorunda değiliz. RxJS için en bilinen ESLint plugin'i Nicholas Jamieson'ın eslint-plugin-rxjs'iydi; npm'deki son sürümü 5.0.3 (Mart 2023) ve peer bağımlılığı yalnız ESLint 8. Bakımı süren fork'u eslint-plugin-rxjs-x: 1.0.6 sürümü Ağustos 2026'da çıktı, ESLint 10 ve typescript-eslint'in flat config'iyle çalışıyor, kural adlarının başında rxjs/ yerine rxjs-x/ var. Peer bağımlılıkları TypeScript'i <6.1.0 ile ve RxJS'i ^7.2.0 ile sınırlıyor; Angular 22.2 projesinin TypeScript 6.0'ıyla uyumlu. Kuralların çoğu tip bilgisine ihtiyaç duyuyor (bir değerin Observable mı, Subject mi olduğunu bilmeleri gerekiyor), bu yüzden typed linting açık olmalı:

// @ts-check
import { defineConfig } from 'eslint/config';
import rxjsX from 'eslint-plugin-rxjs-x';
import tseslint from 'typescript-eslint';

export default defineConfig({
  files: ['src/**/*.ts'],
  extends: [tseslint.configs.recommended, rxjsX.configs.recommended],
  languageOptions: {
    // RxJS kuralları tip bilgisine ihtiyaç duyar: bu bir Observable mı, Subject mi, Subscription mı?
    parserOptions: { projectService: true, tsconfigRootDir: import.meta.dirname },
  },
  rules: {
    // "recommended"da yok, ama bu serinin önem verdikleri
    'rxjs-x/no-ignored-subscription': 'error',
    'rxjs-x/no-exposed-subjects': 'error',
    'rxjs-x/no-floating-observables': 'error',
    // takeUntilDestroyed'ı takeUntil gibi say: ondan sonra yalnız finalize ve benzerleri gelebilir
    'rxjs-x/no-unsafe-takeuntil': ['error', { alias: ['takeUntilDestroyed'] }],
  },
});

npm i -D eslint typescript-eslint eslint-plugin-rxjs-x ile kurduk ve package.json'a "lint": "eslint ." ekledik. recommended config'i no-nested-subscribe, no-unsafe-takeuntil, no-ignored-replay-buffer gibi kuralları açıyor; no-ignored-subscription, no-exposed-subjects ve no-floating-observables'ı biz ekledik. no-unsafe-takeuntil varsayılan olarak yalnız takeUntil'e bakıyor; Angular'ın takeUntilDestroyed'ını da kapsaması için alias seçeneğini verdik. Kuralları denemek için önceki yazılardaki hataları tek bir component'te topladık:

import { HttpClient } from '@angular/common/http';
import { Component, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { ActivatedRoute } from '@angular/router';
import { map, Subject } from 'rxjs';
import { apiUrl } from '../api';
import type { Book } from '../catalog/book';

@Component({ selector: 'app-book-title-logger', template: '' })
export class BookTitleLogger {
  readonly #http = inject(HttpClient);
  readonly #route = inject(ActivatedRoute);
  readonly titleLogged = new Subject<string>();

  constructor() {
    this.#route.paramMap.pipe(takeUntilDestroyed()).subscribe((params) => {
      this.#http.get<Book>(`${apiUrl}/books/${params.get('isbn')}`).subscribe((book) => {
        this.titleLogged.next(book.title);
      });
    });
    this.#route.paramMap
      .pipe(
        takeUntilDestroyed(),
        map((params) => params.get('isbn')),
      )
      .subscribe((isbn) => console.log(`[book] ${isbn}`));
  }

  markAsSeen(isbn: string): void {
    this.#http.post(`${apiUrl}/books/${isbn}/seen`, {});
  }
}

src/app/lint-demo/book-title-logger.ts
  13:12  error  Subject 'titleLogged' must be private  rxjs-x/no-exposed-subjects
  17:69  error  Nested subscribe calls are forbidden  rxjs-x/no-nested-subscribe
  17:69  error  Ignoring returned subscriptions is forbidden  rxjs-x/no-ignored-subscription
  23:9   error  Applying operators after takeUntil is forbidden  rxjs-x/no-unsafe-takeuntil
  26:8   error  Ignoring returned subscriptions is forbidden  rxjs-x/no-ignored-subscription
  30:5   error  Observables must be subscribed to, returned, converted to a promise and awaited, or be explicitly marked as ignored with the `void` operator  rxjs-x/no-floating-observables

✖ 6 problems (6 errors, 0 warnings)

Altı hatanın arkasında bu seride gördüğümüz sorunlar var: dışarıya açık bir Subject (Subject türleri yazısı), iç içe subscribe (önceki yazı), takeUntilDestroyed'dan sonra gelen bir operator ve subscribe edilmediği için hiç gönderilmeyen bir POST. no-unsafe-takeuntil, takeUntil'den sonra gelen bir switchMap'in iç subscription'ının temizlenmemesi riskine karşı yazılmış ve izin listesindekiler (finalize, count, last gibi) dışında her operator'ü işaretliyor. Buradaki map aslında zararsız, ama kural ayırt etmiyor; takeUntilDestroyed'ı sona almak hem güvenli hem okunur. 26. satırdaki hata da bunun sonucu: no-ignored-subscription, son operator take, first, takeUntil ya da takeUntilDestroyed gibi akışı bitiren bir operator ise subscription'ın yok sayılmasına izin veriyor; map sonda olduğu için kural onu bitmeyen bir subscription olarak görüyor.

Gerçek kodda ise tek bir hata çıktı: sepet sayfasındaki placeOrder çağrısı. Kural bir HTTP isteğinin tek yanıttan sonra tamamlandığını bilmiyor. Burada takeUntilDestroyed eklemek yanlış olurdu, çünkü kullanıcı sayfadan çıkınca tarayıcıdaki isteği iptal eder ama sunucudaki siparişi iptal etmez. Doğrusu kuralı o satırda gerekçesiyle kapatmak:

    this.#orders
      .placeOrder(this.cart.items())
      // ne olursa olsun buton geri gelir: başarı da hata da
      .pipe(finalize(() => this.placing.set(false)))
      // eslint-disable-next-line rxjs-x/no-ignored-subscription -- tek bir POST: HttpClient yanıttan sonra tamamlanır
      .subscribe({
        next: (order) => this.message.set(`Order ${order.id} placed.`),
        // component seviyesi: 409'un kullanıcı için ne demek olduğunu bu sayfa biliyor
        error: (err: unknown) => this.message.set(errorMessage(err)),
      });

Lint kuralları sezgisel; yanlış pozitif veriyorlarsa kuralı kapatmak yerine o satırda neden yanlış pozitif olduğunu yaz. -- … ile yazılan açıklama, aylar sonra o satıra gelen birine aynı tartışmayı tekrar yapmamayı sağlıyor.

Üretimde Dikkat

Custom operator'ler uygulamanın küçük bir library'si gibi davranır; bu yüzden onlara library muamelesi yap. Tek bir dosyada (core/operators.ts) dur, her biri kısa bir yorumla ne yaptığını söylesin ve testleri olsun. Bir sonraki yazıda bu operator'leri TestScheduler ile test edeceğiz; retryWithBackoff'un random seçeneği de tam bunun için var.

debugTap'i production build'e taşıma ya da taşıyacaksan log fonksiyonunu ortama göre bağla. Her next'te JSON.stringify çağırmak büyük bir liste dönen bir akışta ölçülebilir bir maliyet. Geliştirme sırasında ise sızıntı ararken en hızlı araçlardan biri: bir sayfadan çıktıktan sonra unsubscribe satırı görmüyorsan bir subscription açık kalmış demektir.

ESLint'i CI'da çalıştır ve typed linting'in maliyetini hesaba kat: tip bilgisi gerektiren kurallar TypeScript'in program'ını kurduğu için düz lint'ten belirgin şekilde yavaş. Bizim küçük projemizde npx eslint . 2,2 saniye sürdü; büyük bir monorepo'da lint'i yalnız değişen proje için çalıştırmak gerekebilir. Son olarak, eslint-plugin-rxjs-x'in peer aralığına bak: RxJS ya da TypeScript sürümünü yükselttiğinde lint adımı ilk kırılan yer olabilir.

Sık Düşülen Tuzaklar

• Sıfırdan yazılan operator'de teardown döndürmemek. Akış bitse de kaynak çalışmaya devam eder; tahmin örneğindeki polling gibi.
• Kaynağın error ya da complete'ini aşağıya iletmemek. Hata kaybolur ya da akış hiç bitmez.
• Generic'i any'ye düşürmek. Yazım hataları derlenir ve çalışma anında undefined olarak ortaya çıkar.
• Hatayı değere çeviren bir operator'ü (loadingState, catchError) retry'dan önce koymak. Retry hiç çalışmaz.
• Bir tap içinde değer değiştirmek ya da yeni bir istek başlatmak. O iş map ya da bir higher-order mapping operator'ünün.
• Mevcut bir operator'ün yaptığı işi sıfırdan yazmak. stockChanges aslında distinctUntilChanged() idi.
• Lint uyarısını sebebini yazmadan kapatmak ya da kuralı bütün projede kapatmak.

Küçük Bir Deneme

★☆☆ debugTap'e bir enabled parametresi ekle; false olduğunda kaynağı hiç sarmadan olduğu gibi döndürsün. Neden new Observable oluşturup içinde log'u atlamaktan daha iyi?

★★☆ retryWithBackoff'a, HTTP yanıtındaki Retry-After header'ını okuyup varsa backoff yerine o süreyi bekleyen bir seçenek ekle. Sahte API'ye 503 ile birlikte Retry-After: 2 döndüren bir yol ekleyip interceptor'da dene.

★★★ Bir değeri en az ms milisaniye ekranda tutan bir minDuration(ms) operator'ü yaz: istek 50 ms'de dönse bile loading durumu 300 ms görünsün, böylece ekran titremesin. Önce mevcut operator'lerle dene (combineLatest ve timer ipucu), sonra sıfırdan yaz ve iki sürümün teardown davranışını debugTap ile karşılaştır.

Bu yazıda operator'lerin davranışını log çıktılarıyla gösterdik. Bir sonraki yazıda aynı davranışı test olarak yazıyoruz: TestScheduler ile marble testing, settledTerm ve retryWithBackoff için sanal zamanda çalışan testler, aynı testlerin Vitest fake timer'larıyla karşılaştırması ve HttpTestingController ile Bookstore'un Angular service'leri.