RxJS’te Tamponlama Operatörlerine Giriş

RxJS, reaktif programlama prensiplerini JavaScript uygulamalarına taşıyan güçlü bir kütüphanedir. Zamanla değişen veya olay tabanlı veri akışlarını (observables) yönetmek için zengin bir operatör seti sunar. Bu operatörler arasında, veri akışını belirli koşullara göre gruplandırarak veya biriktirerek işleyen tamponlama (buffering) operatörleri özel bir yere sahiptir. Bu rehberde, RxJS’in tamponlama operatörlerinin ne olduğunu, nasıl çalıştıklarını ve gerçek dünya senaryolarında nasıl kullanılabileceklerini detaylıca inceleyeceğiz.

RxJS’te Tamponlama Operatörlerine Giriş

RxJS’te tamponlama, bir Observable’dan gelen değerleri belirli bir süre veya belirli bir miktar boyunca biriktirme ve ardından bu birikmiş değerleri tek bir dizi (array) olarak yayma işlemidir. Bu işlem, genellikle kaynak Observable’ın çok hızlı veri üretmesi veya belirli bir zamanda birden fazla olayı tek bir toplu işlem olarak ele almak istediğimiz durumlarda oldukça faydalıdır.

Neden Tamponlama Kullanmalıyız?

• Performans İyileştirme: Her değeri ayrı ayrı işlemek yerine, birden fazla değeri toplu olarak işleyerek uygulama performansını artırabiliriz. Örneğin, her tuş vuruşunda bir sunucu isteği göndermek yerine, belirli bir süre içindeki tüm tuş vuruşlarını toplayıp tek bir istekte gönderebiliriz.
• Veri Akışı Yönetimi: Kaynak Observable’ın hızını ve tüketici Observable’ın işleme hızını dengelemek için kullanılabilir. Bu, “geri basınç” (backpressure) durumlarını yönetmeye yardımcı olabilir.
• API Kısıtlamaları: Belirli bir zaman diliminde yalnızca belirli sayıda API çağrısı yapmanıza izin veren API’lerle çalışırken, istekleri tamponlayıp toplu olarak göndermek kısıtlamalara uymanıza yardımcı olabilir.
• Kullanıcı Etkileşimleri: Çift tıklama algılama, sürükle ve bırak hareketleri gibi birden fazla kullanıcı etkileşimini tek bir olay olarak ele almak için idealdir.

RxJS, bu ihtiyaçları karşılamak için çeşitli tamponlama operatörleri sunar: buffer(), bufferCount(), bufferTime(), bufferToggle() ve bufferWhen(). Şimdi bu operatörleri tek tek inceleyelim.

Temel Tamponlama Operatörleri

buffer()

buffer() operatörü, bir “açma” (notifier) Observable’ı tarafından kontrol edilir. Kaynak Observable’dan gelen değerleri, açma Observable’ı bir değer yayana kadar biriktirir. Açma Observable’ı bir değer yaydığında, mevcut tamponu bir dizi olarak yayar ve yeni bir tampon oluşturmaya başlar.

Nasıl Çalışır?

Kaynak Observable değerler yaymaya başladığında, buffer() bunları bir dahili diziye ekler. Eğer açma Observable’ı bir değer yayarsa, buffer() bu dahili diziyi kaynak Observable’ın abonesine gönderir ve dahili diziyi sıfırlayarak yeni değerleri toplamaya başlar.

import { interval, fromEvent } from 'rxjs';
import { buffer } from 'rxjs/operators';

// Her 500 ms'de bir sayı yayan kaynak Observable
const source = interval(500);

// Her tıklamada tamponu kapatacak açma Observable'ı
const clickBufferNotifier = fromEvent(document, 'click');

const bufferedClicks = source.pipe(
  buffer(clickBufferNotifier)
);

bufferedClicks.subscribe(val => {
  console.log('Tampon kapatıldı, değerler:', val);
});

// Çıktı (Örnek):
// (Kullanıcı 1.5 saniye sonra tıklar)
// Tampon kapatıldı, değerler: [0, 1, 2]
// (Kullanıcı 1 saniye sonra tekrar tıklar)
// Tampon kapatıldı, değerler: [3, 4]

Pratik Bilgiler: buffer(), kullanıcı etkileşimleriyle veya başka harici olaylarla tetiklenen toplu işlemleri yönetmek için mükemmeldir. Örneğin, bir “Kaydet” düğmesine basıldığında, o ana kadar yapılan tüm değişiklikleri tek bir pakette gönderebilirsiniz.

bufferCount()

bufferCount() operatörü, belirli bir sayıda öğe topladıktan sonra tamponu yayar. İsteğe bağlı olarak, tamponların ne kadar sıklıkla açılacağını belirten bir startBufferEvery parametresi de alabilir. Bu parametre, kayan pencereler (sliding windows) oluşturmanıza olanak tanır.

Nasıl Çalışır?

• Sadece bufferSize ile: Kaynak Observable’dan gelen ilk bufferSize kadar öğeyi toplar, tamponu yayar, sonra yeni bir tampon oluşturur ve baştan başlar.
• bufferSize ve startBufferEvery ile: Her startBufferEvery kadar öğe geldiğinde yeni bir tampon oluşturur. Bu, önceki tamponlar kapanmadan yeni tamponların açılmasına neden olabilir.

import { interval } from 'rxjs';
import { bufferCount } from 'rxjs/operators';

const source = interval(1000); // Her saniye bir sayı

// Her 3 öğede bir tampon
source.pipe(
  bufferCount(3)
).subscribe(val => console.log('Her 3 öğede bir:', val));

// Çıktı (Örnek):
// Her 3 öğede bir: [0, 1, 2]
// Her 3 öğede bir: [3, 4, 5]

// Kayan pencere: Her 3 öğede bir tampon, ancak her 1 öğede yeni bir tampon başlat
source.pipe(
  bufferCount(3, 1) // 3 öğe topla, her 1 öğede yeni bir tampon başlat
).subscribe(val => console.log('Kayan pencere (3, 1):', val));

// Çıktı (Örnek):
// Kayan pencere (3, 1): [0, 1, 2]
// Kayan pencere (3, 1): [1, 2, 3]
// Kayan pencere (3, 1): [2, 3, 4]

Pratik Bilgiler: bufferCount(), sayfa numaralandırma, belirli boyutlardaki veri paketlerini işleme veya bir dizi veriyi küçük, yönetilebilir parçalara bölme gibi senaryolarda kullanışlıdır. Kayan pencere özelliği, zaman serisi analizi veya anlık ortalamalar hesaplamak için kullanılabilir.

bufferTime()

bufferTime() operatörü, belirli bir süre boyunca gelen öğeleri toplar ve ardından tamponu yayar. Bu, zaman tabanlı tamponlama için idealdir. İsteğe bağlı olarak, tamponların ne kadar sıklıkla açılacağını belirten bir bufferCreationInterval parametresi de alabilir, bu da kayan pencereler oluşturmanıza olanak tanır.

Nasıl Çalışır?

• Sadece bufferTimeSpan ile: Her bufferTimeSpan süresi dolduğunda, o süre boyunca toplanan öğeleri yayar ve yeni bir tampon başlatır.
• bufferTimeSpan ve bufferCreationInterval ile: Her bufferCreationInterval süresi dolduğunda yeni bir tampon oluşturur. Tamponlar bufferTimeSpan süresi dolduğunda yayılır. Bu, aynı anda birden fazla açık tamponun olmasına neden olabilir.

import { interval } from 'rxjs';
import { bufferTime } from 'rxjs/operators';

const source = interval(500); // Her 500 ms'de bir sayı

// Her 2 saniyede bir tampon
source.pipe(
  bufferTime(2000) // 2000 ms = 2 saniye
).subscribe(val => console.log('Her 2 saniyede bir:', val));

// Çıktı (Örnek):
// Her 2 saniyede bir: [0, 1, 2]
// Her 2 saniyede bir: [3, 4, 5, 6]

// Kayan pencere: 1 saniye süren tamponlar, her 500 ms'de yeni bir tampon başlat
source.pipe(
  bufferTime(1000, 500) // 1 saniye süren tampon, her 0.5 saniyede yeni tampon
).subscribe(val => console.log('Kayan pencere (1000, 500):', val));

// Çıktı (Örnek):
// Kayan pencere (1000, 500): [0, 1]
// Kayan pencere (1000, 500): [1, 2]
// Kayan pencere (1000, 500): [2, 3]

Pratik Bilgiler: bufferTime(), API isteklerini belirli aralıklarla toplu göndermek, oyunlarda belirli bir süre içindeki tüm vuruşları saymak veya gerçek zamanlı veri akışlarında periyodik raporlar oluşturmak gibi senaryolarda kullanılır.

bufferToggle()

bufferToggle() operatörü, tamponun açılma ve kapanma zamanlarını iki ayrı Observable ile kontrol etmenizi sağlar. Bir “açma” Observable’ı, tamponu açar. Her açılma olayı için, bir “kapatma” Observable’ı döndüren bir fonksiyon çağrılır. Bu kapatma Observable’ı bir değer yaydığında, ilgili tampon kapatılır ve yayılır.

Nasıl Çalışır?

Kaynak Observable değerler yayarken, bufferToggle() açma Observable’ını dinler. Açma Observable’ı bir değer yaydığında, yeni bir tampon başlatır ve aynı zamanda kapatma Observable’ı üreten fonksiyonu çağırır. Bu kapatma Observable’ı bir değer yaydığında, o tamponu kapatır ve yeni bir tampon açar.

import { interval, timer } from 'rxjs';
import { bufferToggle } from 'rxjs/operators';

const source = interval(500); // Her 500 ms'de bir sayı

// Açma Observable'ı: Her 3 saniyede bir tampon aç
const openings = interval(3000);

// Kapatma Observable'ı üreten fonksiyon: Her tampon 2 saniye sonra kapanır
const closingSelector = (val: number) => timer(2000);

source.pipe(
  bufferToggle(openings, closingSelector)
).subscribe(val => console.log('Tampon kapatıldı (toggle):', val));

// Çıktı (Örnek):
// (3 saniye sonra açılır, 2 saniye sonra kapanır)
// Tampon kapatıldı (toggle): [0, 1, 2]
// (3 saniye sonra açılır, 2 saniye sonra kapanır)
// Tampon kapatıldı (toggle): [6, 7, 8]

Pratik Bilgiler: bufferToggle(), daha karmaşık, olay tabanlı tamponlama senaryoları için uygundur. Örneğin, bir kullanıcının fareyi bir alana sürüklediğinde tamponu açıp, fareyi alandan çıkardığında kapatmak gibi.

bufferWhen()

bufferWhen() operatörü, bufferToggle()‘a benzer şekilde dinamik tamponlama sağlar, ancak daha basittir. Tamponu kapatmak ve yeni bir tamponu açmak için bir Observable döndüren bir fonksiyon alır. Bu fonksiyon, her tampon kapandığında tekrar çağrılır ve yeni bir kapatma Observable’ı üretir.

Nasıl Çalışır?

Kaynak Observable değerler yayarken, bufferWhen() bir kapatma Observable’ı üretmek için sağlanan fonksiyonu çağırır. Bu kapatma Observable’ı bir değer yaydığında, mevcut tamponu kapatır, yayar ve fonksiyonu tekrar çağırarak yeni bir kapatma Observable’ı ile yeni bir tampon başlatır.

import { interval, timer } from 'rxjs';
import { bufferWhen } from 'rxjs/operators';

const source = interval(500); // Her 500 ms'de bir sayı

// Her tampon 3 saniye sonra kapanır ve yeni tampon başlar
const closingSelector = () => timer(3000);

source.pipe(
  bufferWhen(closingSelector)
).subscribe(val => console.log('Tampon kapatıldı (when):', val));

// Çıktı (Örnek):
// (3 saniye sonra kapanır)
// Tampon kapatıldı (when): [0, 1, 2, 3, 4]
// (3 saniye sonra kapanır)
// Tampon kapatıldı (when): [5, 6, 7, 8, 9]

Pratik Bilgiler: bufferWhen(), tamponun kapanma koşulunun dinamik olarak belirlenmesi gereken durumlarda kullanılır. Örneğin, bir ağ isteğinin tamamlanması veya belirli bir işlem bittiğinde tamponu kapatmak.

Doğru Tamponlama Operatörünü Seçmek

Hangi tamponlama operatörünü seçeceğiniz, veri akışınızı hangi kriterlere göre gruplandırmak istediğinize bağlıdır:

• Sabit Boyutlu Gruplar İçin: bufferCount(). Belirli bir sayıda öğeyi toplamak istediğinizde (örneğin, her 10 veri noktasında bir işlem yap).
• Sabit Zaman Aralıklı Gruplar İçin: bufferTime(). Belirli bir süre boyunca gelen öğeleri toplamak istediğinizde (örneğin, her 5 saniyede bir tüm tıklamaları kaydet).
• Harici Bir Olayla Tetiklenen Gruplar İçin: buffer(). Başka bir Observable’dan gelen bir değerin tamponu kapatmasını istediğinizde (örneğin, bir düğmeye basıldığında).
• Dinamik Açma/Kapama Koşulları İçin: bufferToggle() veya bufferWhen(). Tamponların açılış ve kapanış zamanlarını daha karmaşık, olay tabanlı veya dinamik koşullara göre kontrol etmek istediğinizde. bufferToggle() açma ve kapatmayı ayrı ayrı kontrol ederken, bufferWhen() her kapatmadan sonra yeni bir kapatma Observable’ı üretir.

Gelişmiş Konular ve Pratik Hususlar

Hata Yönetimi

Tamponlama operatörleri, Observable akışındaki hataları genellikle tamponun kapanmasına ve hata içeren akışın sonlanmasına neden olur. Tampon içindeki bir değer hata verirse, bu hata genellikle tüm akışı etkiler. Bu tür durumları yönetmek için catchError operatörünü tamponlamadan önce veya sonra kullanabilirsiniz. Eğer tamponun içindeki bir hata tüm akışı sonlandırmasın istiyorsanız, catchError‘ı tamponlamadan önce uygulamanız gerekebilir, ancak bu durumda hatalı öğenin tampona dahil edilmeyeceğini unutmayın.

Performans ve Bellek Tüketimi

Büyük veya hızlı bir kaynak Observable ile çalışırken tamponlama, bellekte geçici olarak büyük miktarda veri tutulmasına neden olabilir. Özellikle bufferTime() veya bufferCount() kullanırken, tampon boyutlarını (bufferSize veya bufferTimeSpan) dikkatli seçmek önemlidir. Aşırı büyük tamponlar bellek sızıntılarına veya uygulamanızın yavaşlamasına yol açabilir.

Diğer RxJS Operatörleriyle Karşılaştırma

Tamponlama operatörleri, bazen debounceTime() veya throttleTime() gibi operatörlerle karıştırılabilir. Aralarındaki temel fark şudur:

• debounceTime(): Bir Observable’dan gelen değerleri belirli bir süre bekler ve bu süre içinde başka bir değer gelmezse sadece en son değeri yayar. Amacı, hızlı art arda gelen olaylardan yalnızca sonuncuyu almaktır.
• throttleTime(): Bir Observable’dan gelen değeri yayar ve ardından belirli bir süre boyunca gelen diğer değerleri yoksayar. Amacı, olayların yayılma hızını sınırlamaktır.
• Tamponlama Operatörleri: Belirli bir süre veya sayıdaki tüm değerleri toplar ve bunları bir dizi olarak yayar. Amacı, birden fazla değeri toplu olarak işlemektir.

Bu operatörler farklı amaçlara hizmet eder, ancak bazen bir arada kullanılarak daha karmaşık veri akışı senaryoları oluşturulabilir.

Sonuç

RxJS tamponlama operatörleri, reaktif programlamada veri akışını yönetmek için son derece güçlü ve esnek araçlardır. İster belirli sayıda öğeyi, ister belirli bir zaman dilimi içindeki öğeleri, isterse de harici olaylarla tetiklenen öğeleri gruplandırmak isteyin, RxJS’in sunduğu zengin tamponlama operatörleri koleksiyonunda ihtiyacınıza uygun bir çözüm mutlaka bulacaksınız. Bu operatörleri doğru bir şekilde anlamak ve uygulamalarınızda kullanmak, daha verimli, tepkisel ve yönetilebilir kod yazmanıza olanak tanır. Uygulamalarınızda bu operatörleri denemekten ve farklı senaryolarda nasıl davrandıklarını keşfetmekten çekinmeyin.

Sıkça Sorulan Sorular (SSS)

buffer() ile bufferCount() arasındaki temel fark nedir?

buffer() operatörü, tamponun ne zaman kapatılacağını ve yayılacağını harici bir “açma” Observable’ı belirler. Örneğin, bir düğmeye tıklama gibi bir olayla tetiklenir. bufferCount() ise, kaynak Observable’dan belirli sayıda (bufferSize) öğe geldiğinde tamponu kapatır ve yayar. Yani, buffer() olay tabanlı, bufferCount() ise sayı tabanlıdır.

Tamponlar bellek sorunlarına neden olabilir mi?

Evet, özellikle kaynak Observable çok hızlı veri üretiyorsa ve tamponlama operatörü çok uzun bir süre veya çok fazla öğe biriktirecek şekilde yapılandırılmışsa, bellekte geçici olarak büyük miktarda veri tutulmasına neden olabilir. Bu durum, bellek tüketiminin artmasına ve uygulamanızın performansının düşmesine yol açabilir. Bu nedenle, tampon boyutlarını (bufferTimeSpan, bufferSize) dikkatli bir şekilde belirlemek önemlidir.

Tamponlar debounceTime veya throttleTime ile nasıl karşılaştırılır?

debounceTime ve throttleTime operatörleri, bir Observable’dan gelen tekil değerlerin yayılmasını kontrol eder. debounceTime, bir dizi olayın sonuncusunu yayarken, throttleTime belirli bir süre boyunca sadece ilk olayı yayar ve diğerlerini yoksayar. Tamponlama operatörleri ise, birden çok değeri tek bir dizi (array) halinde toplar ve yayar. Yani, debounceTime/throttleTime filtreleme ve hız kontrolü yaparken, tamponlama toplama (batching) işlemi yapar.

Tamponlamadan ne zaman kaçınmalıyım?

Her öğenin anında ve bağımsız olarak işlenmesi gereken durumlarda tamponlamadan kaçınmalısınız. Ayrıca, bellek kısıtlamaları çok darsa ve tampon boyutu üzerinde yeterli kontrol sağlayamıyorsanız, veya debounceTime veya throttleTime gibi daha basit bir filtreleme/hız kontrolü yeterliyse tamponlama gereksiz karmaşıklık getirebilir.