RxJS’e yakından bakış -4: Pipe, Misket Diyagramları ve Operatörler.

By Tahir Kardak Last updated Feb 16, 2021

Merhaba bu yazıda pipe fonksiyonu, operatörlerin daha iyi anlaşılmasında faydalı olacağını düşündüğüm Misket Diyagramları (Marbel Diagrams) ve operatörler hakkında temel bilgilerden bahsedeceğim. Bu yazıdan sonra operatörler hakkında detaylı bir yazı dizisi hazırlayıp yayınlayacağım.

PIPE: Tut şunun ucunu döşeyelim abi.

İçinden su yerine verilerin dönüştürülerek (operatör kullanılarak örneğin, filter, map, vb ) aktığı bir boru gibi düşünebiliriz. Pipe işlemi tamamlandıktan sonra observable’a subscribe(abone) olarak akış başlatılır.

Pipe fonksiyonu, observable veri kaynağındaki, operatörler ile montaj hattıdır. Tıpkı bir arabanın bitmiş halinin fabrikadan çıkmadan önce uğradığı bir dizi üretim bandı gibi düşünebilirsiniz.

Pipe’in bize sunduğu avantajlardan bir kaçını şu şekilde sıralayabiliriz.

RxJS kütüphanelerinin tree-shakeable(ihtiaç duyulmadıkça import edilmeyen ve bundle boyutunu büyütmeyen) olmasını sağlar.
third-party operatör yazmayı ve kullanmayı kolaylaştırır.

Kendi pipe fonksiyonumuz yazarak nasıl çalıştığını daha iyi anlayabiliriz diye düşünüyorum. Aşağıdaki örneği inceleyelim.

StackBlitz

Örneğimizde pipe fonksiyonu rest parametre(...) tipinde olan fonksiyonlar parametresini almaktadır. Daha sonra baslangicDegeri değişkenini parametre olarak alan bir fonksiyon dönüyor. Bu fonksiyon baslangicDegeri değişkenini reduce fonksiyonuna geçiyor. Bu değer fonksiyonlar’in içinde bulunan topla fonksiyonuna parametre olarak geçiyor. Bu fonksiyondan dönen değer, carp fonksiyonuna parametre olarak geçiyor. İşlemin sonunda dönen değer bol fonksiyonuna parametre olarak geçiyor. Yine bu fonksiyondan dönen değer mod fonksiyonuna parametre olarak geçiyor. Eğer fonksiyonlar parametresinin içinde daha fazla fonksiyon bulunsaydı dönen değer sırasıyla diğer fonksiyonlara geçer ve süreç tamamlanırdı. Aslında pipe fonksiyonun yaptığı işi şu şekilde özetleyebiliriz.



const topla = x => x + 2;
const carp = x => x * 5;
const bol = x => x / 2;
const mod = x => x % 6;
const ilkDeger = 100;console.log(mod(bol(carp(topla(ilkDeger)))));

const topla = x => x + 2;

const carp = x => x * 5;

const bol = x => x / 2;

const mod = x => x % 6;

const ilkDeger = 100;console.log(mod(bol(carp(topla(ilkDeger)))));

Dikkat: pipe fonksiyonuna verdiğimiz tüm fonksiyonlar aynı tipte değer dönüyor.

Ayrıca baslangicDegeri’de bir fonksiyondur. Şu şekilde ayrıca tanımlayabiliriz.



function = baslangicDegeri(x) {
return x;
}
yada => ile yazarsak
const baslangicDegeri = x => x;

function = baslangicDegeri(x) {

return x;

}

yada => ile yazarsak

const baslangicDegeri = x => x;

Rxjs içerisinde bulunan pipe fonksiyonu da buna benzer bir mantık ile çalışmaktadır. pipe fonksiyonu kullanılarakObservable dönen pipeable operatörleri(filter, map vb) birleştirerek sonuç döner.

Aşağıdaki örnekte map operatörü ile pipe fonksiyonunu kullanıyoruz.

StackBlitz

Örneğimizde of ile yaratılan observable’nda yayılan değerler pipe fonksiyonu kullanılarak map operatöründen geçirilererek işlem yapılıyor. Daha sonra observable’a abone olunarak sonuç konsola yazdırılıyor.

Misket Diyagramları(Marbel Diagrams):

Soyut olan Rxjs kavramlarını (observer, subcription, operatör vb)daha iyi anlayabilmemiz için tasarlanmış diyagramlardır.

Aşağıdaki örnekte interval ile; saniyede bir yayılan ve ilk dört değeri alınan observable yaratılıyor. Konsola sonuçlar birer saniye arayla 0,1,2,3 şeklinde yazılıyor.



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

interval(100)
     .pipe(take(4))
     .subscribe(console.log);

import { interval } from 'rxjs';

import { take } from 'rxjs/operators';

interval(100)

.pipe(take(4))

.subscribe(console.log);

Bu işlemi görsel olarak anlamak için; marbel diyagramında akışın nasıl göründüğüne bakalım. Denemek için burayı tıklayınız.

Subscribe işlemi yapıldıktan bir saniye sonra 0 değeri, bir saniye sonra 1 değeri, bir saniye sonra 2 değeri, bir saniye sonra 3 değeri yayılıyor. Eğer take operatörünü kullanmasaydık bu akış sonsuza kadar devam edecekti. take operatörüne 4 değerini vererek akışı dördüncü değer alındıktan sonra bitiriyoruz.

Diyagramı sözlü olarak ifade edersek.

Örneğimizdeki soyut olan bir interval işlemi, marbel diyagramı ile daha anlaşılır hale geldi diye düşünüyorum.

Aşağıdaki diyagramda ise filter operatörünü kullanan bir akış görülmektedir. Akış içerisindeki 10’dan büyük olan değerler işlenip observer’a iletiliyor.

Diyagramları kendiniz denemek isterseniz, aşağıdaki iki siteyi tavsiye ederim.

https://rxmarbles.com/

https://rxviz.com/

RxJs Operatörleri

Yazının yıldızlarına geldik. Yıldızlarımızdan bu yazıda detaylı bir şekilde bahsetmeyeceğim. Kendileri hakkında uzun bir yazı serisi hazırlamayı planlıyorum. Takipte kalın.

JavaScript Array metotlarını kullanmış iseniz RxJS operatörleri size tanıdık gelecektir.

Operatörler Observable dönen fonkisyonlardır. İki tür operatör vardır:

Yaratma operatörleri; yeni bir Observable yaratmak için kullanılan bağımsız fonsiyonlardır. Örneğin of(1, 2, 3) ile yaratılan observable 1,2,3 değerlerini yayar.

ObservableInstance.pipe(operator()) söz dizimi ile kullanılan operatörlere pipeable operatör denir. Mevcutta var olan observable’ı değiştirmezler, bunun yerine yeni bir observable dönerler.

Pipeable operatörler; girdi olarak observable alır ve yeni bir observable döner. Önceki observable’ı değiştirmedikleri için pure operatörlerdir.

Pipeable operatörden dönen observable’a abone olunduğunda girdi olarak verilen ilk observable da abone olunur.

Pipeable Operatörleri, observable’dan yayılan değerleri değiştirmemize yardımcı olurlar. Operatörler dönüştürülen değerleri Observable olarak geri dönerler. Bu sayede bir akışta birden çok operatör uygulanabilir.

Aşağıdaki örnekte üç operatör pipe ile art arda uygulanmıştır. Denemek için burayı tıklayınız.



import { interval } from 'rxjs';
import { map, filter,take } from 'rxjs/operators';

interval(700).pipe(
                   map(x => x*5),
                   filter(y=> y>5),
                   take(5))
.subscribe(console.log)

import { interval } from 'rxjs';

import { map, filter,take } from 'rxjs/operators';

interval(700).pipe(

map(x => x*5),

filter(y=> y>5),

take(5))

.subscribe(console.log)

interval kullanılarak yedi yüz mili saniyede bir değer yayan observable yaratılıyor. Akış başladıktan sonra map operatörü ile gelen değer 5 ile çarpılıyor. Daha sonra filter operatörü ile 5 den büyük olan değer filtreleniyor. Bu döngünün sonsuza kadar sürmemesi için take operatörü ile 5 değerden sonra akış tamamlanıyor.

Yukarıdaki görselde ilk değer 1,4 mili saniye sonra yayılıyor. Sebebi ise interval’in yaydığı ilk değer 0 olduğu için, bu değer filter operatörü tarafından eleniyor. Sonra şartlara uyan değerler ekranda gösteriliyor.

Operatör Grupları:

RxJs’te yaptıkları işlere göre operatörler on farklı şekilde gruplandırılabilir.

Yaratma Opeatörleri: Observable yaratmak için kullanılan operatörlerdir.

ajax bindCallback bindNodeCallback defer empty from fromEvent

fromEventPattern generate interval of range throwError timer iif

Birleştirme ve Yaratma Operatöleri: Bu operatörler aynı zamanda birleştirme işlevine sahip Observable yaratan operatörlerdir.

combineLatest concat forkJoin merge race zip

Bu operatörler hakkında detaylı bilgi almak için aşağıdaki yazıyı okumanızı tavsiye ederim

[Çeviri] Süper sezgisel etkileşimli diyagramlar ile RxJs sıralı dizileri birleştirmeyi öğrenin

Dönüşüm operatörleri: Observable’dan yayılan değerleri yeni bir observable içinde dönüştüren operatörlerdir.

buffer bufferCount bufferTime bufferToggle bufferWhen concatMap

concatMapTo exhaust exhaustMap expand groupBy map mapTo mergeMap

mergeMapTo mergeScan pairwise partition pluck scan switchMap

switchMapTo window windowCount windowTime windowToggle windowWhen

Filtreleme Operatörleri: Observable’dan yayılan değerleri filtrelemek için kullanılan operatörlerdir.

audit auditTime debounce debounceTime distinct distinctKey distinctUntilChanged

distinctUntilKeyChanged elementAt filter first ignoreElements last sample

sampleTime single skip skipLast skipUntil skipWhile take takeLast

takeUntil takeWhile throttle throttleTime

Birleştirme Operatörleri: Bu operatörler hakkında detaylı örnek için “Birleştirme ve Yaratma Operatöleri” bölümünde verilen makaleyi okuyabilirsiniz.

combineAll concatAll exhaust mergeAll startWith withLatestFrom

Radyo yayını operatörleri (Multicasting): Bu ismi ben uydurdum ama en güzel tanım olduğunu düşünüyorum.

multicast publish publishBehavior publishLast publishReplay share

Hata Yakalama Operatörleri:

catchError retry retryWhen

Yararlı operatörler:

tap delay delayWhen dematerialize materialize observeOn subscribeOn

timeInterval timestamp timeout timeoutWith toArray

Şartlı ve Boolean Operatörler: defaultIfEmpty every find findIndex isEmpty

Matematiksel ve Kümeleme Operatörleri: count max min reduce

Hangi operatörü ne zaman kullanacağınıza karar vermek için aşağıdaki linki tıklamanızı tavsiye ederim

https://rxjs-dev.firebaseapp.com/operator-decision-tree