RxJS’e yakından bakış -1: Observables

By Tahir Kardak Last updated Eki 20, 2019

Merhaba bu yazımda RxJS’in merkezinde yer alan Observable hakkında daha detaylı bilgi vermeye çalışacağım. Bir önceki yazımda Rxjs’e merhaba demiştik.

Şimdi ise RxJs’e daha yakında bakıp, Çekirdeğinde(Merkezinde) yer alan Observable yakından tanıyacağız.

RxJs dokümanında Observable için şekilde bir açıklama verilmiş:

Observable: Gelecekteki değerler veya olayların bir çalıştırılabilir (invokable) koleksiyon fikrini tarif eder.

Çok açıklayıcı bir tanım olmadığından eminim. Dilimin döndüğü kadar daha anlaşılır bir şekilde ifade etmeye çalışacağım. Dinlenilen bir nesnede meydana gelen değişikliklerin, son durumu hakkında dinleyicilere bildirilmesi işlemi şeklinde bir tanımlama yapabiliriz.

Observable’ı , basitce bir veri kaynağının etrafını saran bir ambalaj kağıdı gibi düşünebiliriz. Veri kaynağını ise bir kaynaktan akan değerler gibi düşünebiliriz. Bu veriler eş zamanlı veya eş zamansız veri kaynağından herhangi zaman diliminden çıkabilir. Gerçek hayattan örnek verecek olursak, Youtube’da Ng Turkey kanalı tarafından yayınlanan efsane, faydalı yeni videoları kaçırmamak için kanala abone olduğumuzu düşünelim. Yayınlanan yeni videoya ilişki bildirimler mail, cep telefonu uygulamasına bildirim veya o an uygulamada(cep telefonu uygulaması, web sitesi vb.) online iseniz ekranda uyarı çanın renginin kızarması şeklinde olur. Burada takip edilen Youtube kanala dinlenilen veri kaynağına Subject(RxJs’teki subject ile karıştırılmasın. Observer paterndeki subjecttir kendisi), kanalı takip eden kişiye Observer , kanala abone olma işlemine Subscribe denir. Observable ise kanalın zaman içerisinde video yayınlamasını temsil eder.

Observerable’da bulunan “subscribe” metodu çağrılmaz(call) ise observerable hiç bir şekilde veri akış işlemini başlatmaz. Yani kanala abone olunmaz ise bildirim gelmez.

Observer’ın tanımını da yaparak eksik parçaları tamamlayalım. Veri akışından gelen veriyi nasıl kullanılacağına karar veren içinde bir tane closed özelliği(property), next, error ve complete metodlarını bulunduran nesneye Observer denir. Obvervable’dan yayılan değer ve uyarıları tüketir.

Observer’ın yapısı şu şekildedir:



interface Observer<T>
{
closed?: boolean;
next: (value: T) => void;
error: (err: any) => void;
complete: () => void;
}

interface Observer<T>

{

closed?: boolean;

next: (value: T) => void;

error: (err: any) => void;

complete: () => void;

}

Observer arayüzü(interface), genellikle observable’da bulunan subscribe metoduna verilir. Observable, bildirimleri sunmak için observser’da bulunan next(value) metodunu çağıracaktır(birden çok kez çağrılabilir). Hatasız işlem durumunda complete() metodunu, hata alırsa error(err) metodunu çağıracaktır. Burada dikkat edilmesi gereken durum complete() metodu kesinlikle parametre almaz.

Observer’ın metot ve özelliklerine yakından bakış.



+----------------+-------------------------------------------------+
| Public         |Members                                          |
+----------------+-------------------------------------------------+
| public         | closed: boolean                                 |
|                | Oberser'ın,Observable'dan abonelik ayrıldığını  |
|                | belirten isteğe bağlı etikettir                 |
+----------------+-------------------------------------------------+

+----------------+-------------------------------------------------+

| Public |Members |

+----------------+-------------------------------------------------+

| public | closed: boolean |

| | Oberser'ın,Observable'dan abonelik ayrıldığını |

| | belirten isteğe bağlı etikettir |

+----------------+-------------------------------------------------+



+----------------+-----------------------------------------------------------+
| Public         |Methods                                                    |
+----------------+-----------------------------------------------------------+
| public         | complete(): void                                          |
|                | Observable tamamlandığında cağrılan callback fonksiyonu   |
+----------------+-----------------------------------------------------------+
| public         | error(err: any): void                                     |
|                | Observable'da hata oluştuğunda, hatanında iliştirildiği   |
|                | callback fonksiyonu                                       |
+----------------+-----------------------------------------------------------+
| public         | next(value: T): void                                      |
|                | Observable'dan gelen bildirim değerlerini alan callback   |
|                | fonksiyonu                                                |
+----------------+-----------------------------------------------------------+

+----------------+-----------------------------------------------------------+

| Public |Methods |

+----------------+-----------------------------------------------------------+

| public | complete(): void |

| | Observable tamamlandığında cağrılan callback fonksiyonu |

+----------------+-----------------------------------------------------------+

| public | error(err: any): void |

| | Observable'da hata oluştuğunda, hatanında iliştirildiği |

| | callback fonksiyonu |

+----------------+-----------------------------------------------------------+

| public | next(value: T): void |

| | Observable'dan gelen bildirim değerlerini alan callback |

| | fonksiyonu |

+----------------+-----------------------------------------------------------+

Makalede yayınlanan örnekleri daha iyi anlayabilmek için Misket Diyagramına bakalım. Bu diyagram Observable’in nasıl çalıştığını gösterir.

Aşağıdaki örnekte. 1,2,3 değerlerini eş zamanlı, 4 değerini süreç başladıktan iki saniye sonra iten ve son itme işlemi bittikten sonra tamamlanan bir observable mevcut.

Denemek için tıklayınız

Örneğimizde hiç bir zaman error metodu çalışmayacaktır. Observable’da tüm değerler tamamlandıktan sonra complete metodu çağrılacaktır. error ve complete metotlarının kullanım sırası önemlidir. Bu metotlardan sonra çağrılan diğer metotlar çalışmayacaktır. Aşağıdaki örnekte bunu görebilirsiniz.

Denemek için tıklayınız

Observerable’ın Tomografisini Çekiyoruz.

Observable birden çok değerin lazy Push koleksiyonlarıdır. Aşağıdaki tabloda eksik kalan noktayı doldurur. Javascript function, iterator ve promise’yi sağlamaktadır.



+------+----------+------------+
|      |  SINGLE  |  MULTIPLE  |
+------+----------+------------+
| Pull | Function | Iterator   |
| Push | Promise  | OBSERVABLE |
+------+----------+------------+

+------+----------+------------+

| | SINGLE | MULTIPLE |

+------+----------+------------+

| Pull | Function | Iterator |

| Push | Promise | OBSERVABLE |

+------+----------+------------+

Pull ve Push İletişim Kuralları

Pull ve Push, bir veri üreticisinin(Producer) bir veri tüketicisiyle(Consumer) nasıl iletişim kurabileceğini açıklayan iki farklı iletişim kuralıdır.

Pull iletişim kuralında; Tüketici, veri kayağından ne zaman veri alacağını belirler. Veri Üreticisi, Tüketiciden habersiz veri yaymaz.

Bütün javascript fonksiyonları Pull iletişim kuralları ile çalışır. Fonksiyon verinin üreticisidir. Fonksiyonu çağırarak fonksiyonun döndüğü değer tüketilir.

ES2015 ile generator functions and iterators (function*) adında başka bir pull sistemi javascripte eklendi.Bu yöntem ile tüketici, veri kaynağından birden çok veriyi çekebilmek için, iterator.next() metodunu çağırır.

Push iletişim kuralında; Veri üreticisi, verinin tüketiciye ne zaman iletileceğine karar verir. Tüketici verinin ne zaman geleceğinden habersizdir.

Javascripte, Promises en çok kullanılan push yöntemidir. Bir Promise (üretici), kayıtlı geri çağrılara (tüketiciler) çözülmüş bir değer sunar.

RxJS Observable ise Javascript için yeni bir push yöntemidir. Bu yöntemde veri üreticisi olan Observable, birden çok verinin Tüketicilere(Observers) gönderilmesini sağlar.



+----------+------------------------------------------+----------------------------------------------+
|          |           PRODUCER(ÜRETİCİ)              |         CONSUMER(TÜKETİCİ)                   |
+----------+------------------------------------------+----------------------------------------------+
| Pull     | pasif: Talep geldiğinde veriyi gönderir. | aktif: Verinin ne zaman alınacağını belirler.|
| Push     | aktif: Veri kendi tempsona göre sunar.   | pasif: Veri kendisine gelince tepki verir.   |
+----------+------------------------------------------+----------------------------------------------+

+----------+------------------------------------------+----------------------------------------------+

| | PRODUCER(ÜRETİCİ) | CONSUMER(TÜKETİCİ) |

+----------+------------------------------------------+----------------------------------------------+

| Pull | pasif: Talep geldiğinde veriyi gönderir. | aktif: Verinin ne zaman alınacağını belirler.|

| Push | aktif: Veri kendi tempsona göre sunar. | pasif: Veri kendisine gelince tepki verir. |

+----------+------------------------------------------+----------------------------------------------+

Fonksiyonların genelleştirilmesi olarak Observable

Observable’lar, parametre almayan fonksiyonlar gibidir, ancak birden çok değere izin vermek için bunları genelleştirir. Aşağıdaki iki örneği inceleyelim. İlk örnekte klasik Javascript fonksiyonu ve çalıştırılması, ikinci örnekte ise RxJs Observable ve çalıştırılması mevcut.

Denemek için tıklayınız

Her iki örnekte bulunan fonksiyonlar tembel(lazy) hesaplama fonksiyonlarıdır. Eğer foo fonksiyonu çağrılmazsa console.log(“Hello”) çalışmayacaktır.

Observableda, Subscribing işlemi bir fonksiyonu çağırmakla (call()) eş değerdir.

İlk örnekteki foo fonksiyonu tamamıyla eş zamanlı bir fonskiyondur. RxJs’te observable’lar hem eş zamanlı hem de ez zamansız olabilir. İkinci örneğimizde 100 ve 200 değerleri eş zamanlı olarak 300 değeri iki saniye sonra eş zamansız olarak yayılmaktadır. Bu yüzdendir ki çıktımızda sonra değeri konsola“300 değerinden önce yazılıyor.

Observable hem eş zamanlı, hemde eş zamansız değer yayabilir

Normal fonksiyon ile Observable arasındaki en büyük fark, Observable zamanla birden çok değer yayabilir. Fonskiyonlar sadece bir değer dönebilir. İlk örneğimizde bulunan foo fonksiyonunda return 100; satırı hiç bir zaman çalışmaz. İkinci örneğimizde ise “next” fonksiyonunu birden çok çağırarak Observable’in birden çok değer dönmesi sağlanabilir.

Özetle

func.call(): Eş zamanlı bir tane değer ver demek
observable.subscribe(): Eş zamanlı veya eş zamansız olarak birden çok değer ver demektir.

Observable’ın anatomisi:

Observable’lar new Observable şeklinde veya bir Observer ile abone olunarak yaratılır.(next / error / complete fonksiyonları çalıştırılarak Observer’a bildirimler gönderilir)

Temelde ilgilendiğimiz konular:

Observable yaratmak
Observable’a abone olmak
Observable’ı çalıştırmak
Observable’ı bırakmak

Observable yaratmak:

Observable constructor’ı bir değer alır. Bu değer ise subscription fonksiyonudur. Aşağıdaki örnekte her bir saniye bir hi değerini yayan bir observable yaratılıyor.



import { Observable } from 'rxjs';const observable= new Observable(function subscribe(subscriber) {
const id = setInterval(() => {
subscriber.next('hi')
}, 1000);
});

import { Observable } from 'rxjs';const observable= new Observable(function subscribe(subscriber) {

const id = setInterval(() => {

subscriber.next('hi')

}, 1000);

});

Observable’lar genellikle new Observable ile yaratılır.RxJS’te bulunan built-in fonksiyonlar ile hemen hemen herşeyden Observable yaratabilirsiniz. En çok kullanılan fonksiyonlar: of, from, interval ve fromEvent. Örnek için burayı tıklayın

Observable’a abone olmak:

Observable’lar subscribe işlemi yapılmadıkça herhangi bir değer yaymaz. Observable’a aşağıdaki şekilde abone olunur.



observable.subscribe(x => console.log(x));

observable.subscribe(x => console.log(x));

Observable’ı çalıştırmak:

Örneğimizde bulunan new Observable(function subscribe(subscriber) {...}) bölümü, observable’ın nasıl çalıştırılacağının tanımlandığı kısımdır. Çalıştırma işlemiyle birlikte eş zamanlı veya eş zamansız olarak değerler yayılabilir.

Observable çalışma süreci boyunca üç farklı tipte değerler yayar:

Next bildirimi: Bir nesne , bir sayı yada string gönderir
Error bildirimi: Hata durumunda hatayı gönderir.
Complete bildirimi: Bir değer göndermez.

Next tipi en önemli ve yaygın olanıdır. Aboneye yayılmaya başlayan gerçek veriyi sunar. Birden çok çağrılabilir. Error ve Complete ise yalnızca bir kez kullanılır.



import { Observable } from 'rxjs';const observable = new Observable(function subscribe(subscriber) {
try {
subscriber.next(1);
subscriber.next(2);
subscriber.next(3);
subscriber.complete();
} catch (err) {
subscriber.error(err);
}
});

import { Observable } from 'rxjs';const observable = new Observable(function subscribe(subscriber) {

try {

subscriber.next(1);

subscriber.next(2);

subscriber.next(3);

subscriber.complete();

} catch (err) {

subscriber.error(err);

}

});

Observable’ı bırakmak:

Observable’lar sonsuza kadar çalışabilir. Ve bu çalışma şekli istenilen bir şey değildir. İşi biten observable’ler bırakılmalı aksi durumda stack overflow hatası alınabilir. fromEvent ile yaratılan tıklama observable’ı ve setIntervalile yaratılan observable sonsuza kadar çalışan bir observable’dır. Örnek için burayı tıklayınız. unsubscribe() fonksiyonu kullanarak observable’lar bırakılır.



import { from } from 'rxjs';const observable = from([10, 20, 30]);
const subscription = observable.subscribe(x => console.log(x));
// Later:
subscription.unsubscribe();

import { from } from 'rxjs';const observable = from([10, 20, 30]);

const subscription = observable.subscribe(x => console.log(x));

// Later:

subscription.unsubscribe();

KAYNAK:

RxJS