Observable

Dalam ReactiveX sebuah observer melakukan subscribe terhadap sebuah Observable. Kemudian observer itu akan bereaksi terhadap apapun item atau urutan dari item-item yang dipancarkan oleh Observable tersebut. Pola seperti ini memfasilitasi operasi-operasi yang berjalan secara konkuren ( berbarengan ) dikarenakan tidak perlu untuk memblokir ketika menunggu Observable untuk memancarkan obyek-obyek, tetapi cara seperti ini malah membentuk sebuah penjaga dalam bentuk sebuah observer yang siap bereaksi terhadap apapun pada waktu kapanpun di masa depan ketika Observable melakukannya.

Halaman ini menjelaskan apa itu pola reactive dan apa sebenarnya Observable dan observer itu (dan bagaimana observer melakukan subscribe kepada Observable). Halaman lainnya menunjukkan bagaimana anda menggunakan berbagai jenis dari operator Observable untuk menghubungkan beberapa Observable dan mengubah perilaku mereka.

Dokumentasi ini akan ditemani oleh “marble diagrams.” Berikut adalah bagaimana diagram kelereng merepresentasikan Observables dan perubahan dari Observables:

Lihat Juga

Single — Sebuah versi Observable yang dibuat khusus untuk memancarkan tepat satu buah item
Rx Workshop: Introduction
Introduction to Rx: IObservable
Mastering observables (from the Couchbase Server documentation)
2 minute introduction to Rx oleh Andre Staltz (“Anggap sebuah Observable sebagai sebuah array asynchronous yang tidak dapat diubah.”)
Introducing the Observable by Jafar Husain (JavaScript Video Tutorial)
Observable object (RxJS) by Dennis Stoyanov
Turning a callback into an Rx Observable by @afterecho

Latar Belakang

Dalam berbagai tugas dalam pemrograman perangkat lunak, instruksi yang anda tulis akan dikerjakan secara bertahap, satu-per-satu, dan berurutan sesuai apa yang anda tulis. Tetapi dalam ReactiveX, banyak instruksi bisa saja dijalankan secara paralel dan hasilnya akan diambil nantinya, dalam suatu urutan yang acak, oleh "observers." Daripada memanggil sebuah method, anda mendefinisikan sebuah mekanisme untuk mendapatkan dan mengubah data-nya, dalam sebuah bentuk "Observable," dan kemudian membuat sebuah observer melakukan subscribe kepadanya, yang dimana mekanisme yang sudah didefinisikan sebelumnya melakukan aksi-nya dengan observer tersebut berdiri sebagai penjaga untuk menangkapnya dan merespon kepada pancarannya kapanpun mereka siap.

Keuntungan dari pendekatan ini adalah ketika anda memmiliki berbagai tugas yang tidak tergantung satu sama lain, anda bisa memulai mereka pada saat yang sama daripada harus menunggu satu per satu dari mereka untuk menyelesaikan tugasnya sebelum memulai tugas berikutnya An advantage of this approach is that when you have a bunch of tasks that are not dependent on each other, you can start them all at the same time rather than waiting for each one to finish before starting the next one — dengan cara begitu, kumpulan tugas anda hanya akan memakan waktu paling lama setara dengan tugas yang memerlukan waktu paling banyak untuk diselesaikan.

Ada beberapa istilah yang digunakan untuk mendeskripsikan model pemrograman dan desain asynchronous. Dokumen ini akan menggunakan istilah berikut: Sebuah observer melakukan subscribe kepada sebuah Observable. Sebuah Observable memancarkan item-item atau mengirim notifikasi-notifikasi kepada observer-nya dengan memanggil method-method dari observer tersebut.

Dalam dokumen dan konteks yang lain, apa yang kita sebut sebagai "observer" kadang-kadang juga disebut sebagai "subsriber," "watcher," atau "reactor." Model seperti ini secara umumnya disebut juga sebagai “reactor pattern”.

Pembuatan Observer

Halaman ini menggunakan pseudocode yang mirip seperti Groovy untuk contoh-contohnya, tetapi ada banyak implementasi ReactiveX dalam bahasa lain juga.

Dalam sebuah pemanggilan method pada biasanya — yang dimana, tidak berjenis pemanggilan asynchronous , dan paralel seperti di ReactiveX — alurnya akan berjalan seperti berikut:

Panggil sebuah method.
Menyimpan hasil dari method tersebut ke dalam sebuah variable.
Menggunakan variable itu dan nilai-nilai nya yang baru untuk sesuatu yang berguna.

Atau, sesuatu seperti ini:

// melakukan pemanggilan, menetapkan hasilnya ke `returnVal`
returnVal = someMethod(itsParameters);
// melakukan sesuatu yang berguna dengan returnVal

Pada model asynchronous, alurnya akan seperti ini:

Mendefinisikan sebuah method yang melakukan sesuatu yang berguna dengan hasil yang dikembalikanya dari sebuah pemanggilan asynchronous; method seperti ini adalah bagian dari observer.
Mendefinisikan pemanggilan asynchronous itu sendiri sebagai sebuah Observable.
Melekatkan observer ke Observable tersebut dengan cara melakukan subscribe kepadanya (dimana ini juga memulai aksi-aksi dari Observable).
Lanjutkan pekerjaan anda; kapanpun panggilan tersebut sudah mengembalikan hasil, method dari observer tersebut akan memulai beroperasi pada nilai tersebut - yaitu item yang diemit oleh Observable tersebut.

Dimana akan mirip sesuatu seperti ini:

// mendefinisikan, tetapi tidak menjalan fungsi onNext dari Subscriber
// (dalam contoh ini, observer yang digunakan sangat sederhana dan hanya mempunyai sebuah fungsi onNext)
def myOnNext = { it -> do something useful with it };
// mendefinisikan, tetapi tidak menjalankan, Observable-nya
def myObservable = someObservable(itsParameters);
// Subsriber melakukan subscribe ke Observable, dan menjalankan fungsi 
myObservable.subscribe(myOnNext) dari Observable;
// lanjutkan hal yang anda perlu kerjakan

onNext, onCompleted, and onError

Subscribe adalah cara agar anda bisa menghubungkan sebuah observer dengan sebuah Observable. Observer anda menerapkan beberapa how you connect an observer to an Observable. Your observer implements some bagian dari method-method berikut:

onNext: Sebuah Observable memanggil method ini kapanpun Observable tersebut memancarkan sebuah item. Method ini memerlukan sebuah parameter item yang dipancarkan oleh Observable.
onError: Sebuah Observable memanggil method ini untuk menandakan bahwa terjadi kegagalan untuk membentuk data yang diharapkan atau telah terjadi error tertentu. Observable tidak akan memanggil onNext atau onCompleted lagi. Method onError memerlukan parameter sesuatu yang menandakan apa yang menyebabkan error tersebut.
onCompleted: Sebuah Observable memanggil method ini setelah sudah memanggil onNext untuk terakhir kalinya, jika belum terjadi error apapun.

Dengan ketentuan dari Kontrak Observable, Observable bisa memanggil onNext sebanyak beberapa kali atau tidak sama sekali, dan kemudian bisa melanjutkan memanggil baik onCompleted atau onError tetapi tidak keduanya, yang dimana akan menjadi panggilannya yang terakhir. Dengan perjanjian, dalam dokumen ini, pemanggil ke onNext biasanya disebut “pancaran” dari item-item, dimana pemanggilan ke onCompleted atau onError disebut “notifikasi.”

Berikut merupakan contoh dari sebuah pemanggilan subscribe yang lebih lengkap:

def myOnNext     = { item -> /* lakukan sesuatu dengan item */ };
def myError      = { throwable -> /* bereaksi terhadap suatu pemanggilan yang gagal */ };
def myComplete   = { /* membersihkan respon terakhir */ };
def myObservable = someMethod(itsParameters);
myObservable.subscribe(myOnNext, myError, myComplete);
// lanjut mengerjakan proses bisnis

Lihat Juga

Introduction to Rx: IObserver

Proses Unsubscribe

Dalam beberapa implementasi ReactiveX, ada sebuah interface obsever khusus, Subscriber, yang mengimplementasi sebuah method unsubscribe. Anda bisa memanggil method ini untuk menunjukkan bahwa Subsciber tersebut tidak lagi tertarik dengan Observable yang sedang dia subscribe. Observable-observable tersebut kemudian (jika mereka tidak mempunyai observer lain yang masih tertarik kepada mereka) berhenti membuat item-item baru untuk dipancarkan.

Hasil dari unsubscription ini akan mengalir kembali melalui rantai operator-operator yang berlaku ke Observable yang disubscribe oleh observer tersebut, dan ini akan menyebabkan masing-masing instansi didalam rantai untuk berhenti memancarkan item. Walaupun tidak dijamin akan langsung terjadi, tetap mungkin bagi sebuah Observable untuk membuat dan mencoba untuk memancarkan item selama beberapa saat bahkan setelah tidak ada observer yang mengamati emisi-emisi tersebut.

Beberapa Catatan tentang Aturan Penamaan

Tiap implementasi yang spesifik kepada bahasa tertentu dari ReactiveX memiliki penamaannya sendiri. Tidak ada aturan penamaan yang standar, meskipun banyak kesamaan diantara implementasi-implementasi tersebut.

Lebih lanjut, beberapa dari nama tersebut memiliki implementasi yang berbeda di konteks lainnya, yang bisa kelihatan seperti kata yang canggung dalam kosa kata bahasa tertentu yang menerapkannya.

Contohnya ada pola penamaan onEvent (contohnya onNext, onCompleted, onError). Dalam beberapa konteks, nama-nama tersebut merupakan method yang memiliki arti penanganan event yang terdaftar. Dalam ReactiveX, bagaimanapun, mereka menamai penanganan event tersebut sendiri.

Observable “Hot” dan “Cold”

Kapan sebuah Observable mulai memancarkan untaian dari item? Tergantung pada Observable itu sendiri. Sebuah Obsevable “hot” bisa mulai memancarkan item segera setelah dibentuk, sehingga observer apapun yang akan melakukan subscribe ke Observable tersebut bisa saja sedang melakukan observasi untaian tersebut di bagian tengah dari untaian tersebut. Di sisi lainnya, Sebuah Observable “cold”, menunggu sampai terdapat sebuah observer yang melakukan subscribe kepadanya sebelum mulai memancarkan item-item, jadi observer tersebut dijamin akan mendapatkan seluruh untaian dari awal.

Pada beberapa implementasi dari ReactiveX, ada sesuatu yang juga disebut sebagai sebuah Observable “Connectable”. Observable tersebut tidak akan mulai memancarkan item sampai method Connect nya dipanggil, tidak peduli ada observer yang melakukan subscribe kepadanya atau tidak.

Komposisi melalui Operator Observable

Observable dan observer merupakan permulaan dari ReactiveX. Mereka tidak lebih dari ekstensi observer pattern, yang lebih cocok untuk mengurus urutan dari suatu event daripada hanya sebuah callback.

Kekuatan sesungguhnya berasal dari "reactive extensions" (asal nama "ReactiveX") - operator-operator yang memungkinkan anda untuk melakukan perubahan, menggabungkan, memanipulasi, dan bekerja dengan untaian item-item yang dipancarkan oleh Observable.

Operator Rx tersebut memungkinkan anda untuk membentuk sebuah untaian asynchronous dengan cara declarative dengan semua keuntungan efisiensi dari callbacks tanpa kelemahan dari penanganan callback yang berasarang yang biasanya diasosiasikan dengan sistem asynchronous.

Dokumentasi berikut menggabungkan informasi tentang berbagai jenis operator dan contoh-contoh penggunaannya ke dalam beberapa halaman berikut:

Membuat Observable: Create, Defer, Empty/Never/Throw, From, Interval, Just, Range, Repeat, Start, dan Timer
Merubah Item Observable: Buffer, FlatMap, GroupBy, Map, Scan, dan Window
Menyaring Observable: Debounce, Distinct, ElementAt, Filter, First, IgnoreElements, Last, Sample, Skip, SkipLast, Take, dan TakeLast
Menggabungkan Observable: And/Then/When, CombineLatest, Join, Merge, StartWith, Switch, dan Zip
Operator Penanganan Error: Catch dan Retry
Beberapa Operator Lainnya: Delay, Do, Materialize/Dematerialize, ObserveOn, Serialize, Subscribe, SubscribeOn, TimeInterval, Timeout, Timestamp, dan Using
Operator Kondisional dan Boolean: All, Amb, Contains, DefaultIfEmpty, SequenceEqual, SkipUntil, SkipWhile, TakeUntil, dan TakeWhile
Operator Matematika dan Agregasi: Average, Concat, Count, Max, Min, Reduce, dan Sum
Mengubah Observable: To
Operator Obervable Connectable: Connect, Publish, RefCount, dan Replay
Operator Backpressure: jenis dari operator-operator yang menerapkan aturan flow-control tertentu

Halaman-halaman tersebut juga berisi informasi mengenai beberapa operator yang tidak termasuk operator inti dari ReactiveX tetapi diimplementasikan di satu atau lebih dari bahasa atau modul tersebut.

Menggabungkan Operator

Kebanyakan operator beroperasi pada sebuah Observable dan mengembalikan sebuah Observable. Ini memungkinkan anda untuk mengaplikasikan operator-operator tersebut satu demi satu, seperti sebuah rantai. Masing-masing operator pada rantai tersebut mengubah Observable yang merupakan hasil dari operasi yang dilakukan operator sebelumnya.

Ada beberapa pattern yang lain, seperti Builder Pattern, yang dimana beberapa method dari sebuah class tertentu beroperasi pada sebuah item pada class yang sama dengan cara mengubah obyeknya melalui operasi method tersebut. Pattern seperti itu juga memungkinkan anda untuk menggabungkan method-method dengan cara yang hampir sama. Tetapi di Builder Pattern, urutan dari method apa yang dipakai di rantai pemanggilan method biasanya tidak berpengaruh, sedangkan pada operator Observable, urutan berpengaruh.

Sebuah rantai operator Observable tidak beroperasi secara indenpenden terhadap Observable asalnya yang memulai rantai tersebut, tetapi beroperasi secara bergantian, masing-masing beroperasi pada Observable yang dibentuk oleh operator sebelumnya dalam rantai tersebut.