如果您寫 Angular, 那您一定對 Observable 不陌生, 當您對後端做 GET, POST, PUT, DELETE 要求時所使用的 HttpClient 就是 Observable. 因為整個 ReactiveX 非常龐大, 甚至包含了其他程式語言, 這裡我們只簡單的介紹針對 Javascript RxJS 而且在 ngrx/store 常用到的基本定義以及運算子, 完整的 RxJs 請參考 RxJS API
產生 Observable
Observable 簡單來說就是資料流, 至於如何產生 Observable 呢? 一般而言有幾種情況
同步靜態的如 純量, Array, Set 或是 Object
Rx.Observable.of(true);
Rx.Observable.from([1,2,3,4]);
const user = [
{ name: ‘John’, age: 36},
{ name: ‘Jason’, age: 24},
{ name: ‘Jennifer’, age: 33}
]
Rx.Observable.from(user);
同步動態的如利用 Event, 例如 mouse event, input event 或是非同步時間等等
Rx.Observable.fromEvent(mouse, ‘mousemove’);
Rx.Observable.interval(1000);
從非同步的 Promise 產生
// getUser from github
function getUser(username) {
return $.ajax({
url: ‘
https://api.github.com/users/
‘ + username,
dataType: ‘jsonp’
}).promise();
}
Rx.Observable.fromPromise(getUser(‘myname’));
自行建立 Observable
new Rx.Observable(observer => {
observer.next(1);
observer.next(2);
observer.next(3);
setTimeout(() => {
observer.next(4);
observer.complete();
}, 1000);
});
Data Provider, Observable, Subscriber 跟 Observer
有幾個比較重要的基本定義
Data Provider: 也就是資料流的來源, 像上面的 of, from, fromPromise… 的對象
Observable: 就是觀察資料流的物件, 它有一個重大的任務, 當資料來源變更時, 需要馬上告訴他的客戶 (Observer, Subscriber) 我的資料變動了, 這個是稱為 push 的方式, 而 push 的方式就是當資料變動時啟動 Observer next 的 method, 所以客戶自己要定義接到這個 next 時， onNext callback 要做什麼因應, 接到 error 時 onError 要做什麼…
Observer: 也就是 Observable 的客戶，主要任務是定義一些 callback 處理當 Observable 啟動 next, error 或是 complete 時要做的事
Subscriber 或 Subscription: 是 Observer 的實作, 除了定義 Observer 要做的 callback 外，另外還可以隨時做 unsubscribe, 而這也是跟 Promise 很不同的地方
public abstract class Subscriber<T> implements Observer<T>, Subscription
用一個例子來說明
const source$ = new Rx.Observable(observer => {
observer.next(1);
observer.next(2);
observer.next(3);
observer.error(new Error(‘error: error message’));
setTimeout(() => {
observer.next(4);
observer.complete();
}, 1000);
});
source$.subscribe(
val => {
console.log(val);
},
err => {
console.log(err);
},
complete => {
console.log(‘completed’);
}
);
codepen
Data Provider: 是簡單的 1, 2, 3
Observable: 也就是 source$, 他要告訴 Observer 資料的變化, 像是 observer.next(1), …
Subscription: 定義了相對應的個別 callback
所以以上的程式會產生
var Observable = (function (subscriber) {
this._subscriber = subscriber;
})
Observable.prototype.subscribe = function(next, error, complete) {
this._subscriber(next);
}
const source$ = new Observable(observer => {
observer(1);
observer(2);
observer(3);
});
const subs = source$.subscribe(
(v) => {
console.log(‘subscribed: ‘ + v)
},
(err) => {
console.log(‘error with’ + err)
},
(complete) => {
console.log(‘complete’);
}
)
codepen
輸出的結果會是
最後談一下 Subject, 因為 ngrx/store 裡的 store 跟 dispatcher 都是 Subject (BehaviorSubject)
Subject 是 Observable 跟 Observer 的延伸, 也就是同時具有 Observable 跟 Observer 的特徵, 他可以被 Subscriber subscribe 也同時 subscribe 到其他的 Observable. 我們可以當它是代理人, 舉個例子
var subject$ = new Rx.Subject();
// acting as Observable
const sub1 = subject$.subscribe({
next: (v) => console.log(‘observerA: ‘ + v)
});
const sub2 = subject$.subscribe({
next: (v) => console.log(‘observerB: ‘ + v)
});
// acting as Observer
subject$.next(1);
subject$.next(2);
codepen
BehaviorSubject
BehaviorSubject 延伸 Subject 再加上它會保留一下最後的一筆資料，也就是即使過了資料產生的時間，後來的訂閱者還是可以拿到最後產生時的資料，也因為如此，在產生新的 BehaviorSubject 時，需要給初值。
如上面例子
var subject$ = new Rx.Subject();
// acting as Observable
const sub1 = subject$.subscribe({
next: (v) => console.log(‘observerA: ‘ + v)
});
const sub2 = subject$.subscribe({
next: (v) => console.log(‘observerB: ‘ + v)
});
// acting as Observer
subject$.next(1);
subject$.next(2);
const sub3 = subject$.subscribe({
next: (v) => console.log(‘observerC: ‘ + v)
});
codepen
sub3 不會得到更新
但是換成 BehaviorSubject
var subject$ = new Rx.BehaviorSubject(0);
// acting as Observable
const sub1 = subject$.subscribe({
next: (v) => console.log(‘observerA: ‘ + v)
});
const sub2 = subject$.subscribe({
next: (v) => console.log(‘observerB: ‘ + v)
});
// acting as Observer
subject$.next(1);
subject$.next(2);
const sub3 = subject$.subscribe({
next: (v) => console.log(‘observerC: ‘ + v)
});
codepen
sub3 還是會拿到最後的值 (2)，產生 ObservableC: 2
有了 Observable 的基本認識, 接下來我們來談一些運算子 Operators